Module livekit.agents.inference

def is_duplex_model(model: str) -> bool:
    """Whether a LiveKit Inference model string names a full-duplex voice model.

    Only the models listed in ``DuplexModels`` count, as with :func:`is_realtime_model`.
    """
    return model in _DUPLEX_MODEL_IDS

def is_realtime_model(model: str) -> bool:
    """Whether a LiveKit Inference model string names a realtime (speech-to-speech) model.

    Only the models listed in ``RealtimeModels`` count: a newly released realtime
    model has to be added there before ``llm="..."`` resolves it to a
    ``RealtimeModel``.
    """
    return model in _REALTIME_MODEL_IDS

def llm_from_model_string(model: str) -> llm.LLM | llm.RealtimeModel | llm.DuplexModel:
    """Create the inference model a ``llm=`` string names.

    Duplex model strings resolve to :class:`livekit.agents.inference.DuplexModel`,
    realtime (speech-to-speech) model strings to
    :class:`livekit.agents.inference.RealtimeModel`, every other string to
    :class:`livekit.agents.inference.LLM`.
    """
    if is_duplex_model(model):
        # imported lazily: the realtime package imports this module
        from .realtime import DuplexModel

        return DuplexModel.from_model_string(model)

    if is_realtime_model(model):
        # imported lazily: the realtime package imports this module
        from .realtime import RealtimeModel

        return RealtimeModel.from_model_string(model)

    return LLM.from_model_string(model)

class AdaptiveInterruptionDetector(
    rtc.EventEmitter[
        Literal[
            "overlapping_speech",
            "error",
            "metrics_collected",
        ]
    ],
):
    def __init__(
        self,
        *,
        threshold: NotGivenOr[float] = NOT_GIVEN,
        min_interruption_duration: float = MIN_INTERRUPTION_DURATION,
        max_audio_duration: float = MAX_AUDIO_DURATION,
        audio_prefix_duration: float = AUDIO_PREFIX_DURATION,
        detection_interval: float = DETECTION_INTERVAL,
        inference_timeout: float = REMOTE_INFERENCE_TIMEOUT,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        """
        Initialize a AdaptiveInterruptionDetector instance.

        Args:
            threshold (float, optional): The threshold for the interruption detection. When not set, the server-recommended default (returned in session.created) is used.
            min_interruption_duration (float, optional): The minimum duration, in seconds, of the interruption event, defaults to 50ms.
            max_audio_duration (float, optional): The maximum audio duration, including the audio prefix, in seconds, for the interruption detection, defaults to 3s.
            audio_prefix_duration (float, optional): The audio prefix duration, in seconds, for the interruption detection, defaults to 0.5s.
            detection_interval (float, optional): The interval between detections, in seconds, for the interruption detection, defaults to 0.1s.
            inference_timeout (float, optional): The timeout for the interruption detection, defaults to 1 second.
            base_url (str, optional): The base URL for the interruption detection, defaults to the shared LIVEKIT_INFERENCE_URL environment variable.
            api_key (str, optional): The API key for the interruption detection, defaults to the LIVEKIT_INFERENCE_API_KEY environment variable.
            api_secret (str, optional): The API secret for the interruption detection, defaults to the LIVEKIT_INFERENCE_API_SECRET environment variable.
            http_session (aiohttp.ClientSession, optional): The HTTP session to use for the interruption detection.
        """
        super().__init__()
        if max_audio_duration > 3.0:
            raise ValueError("max_audio_duration must be less than or equal to 3.0 seconds")

        lk_base_url = base_url if base_url else get_default_inference_url()

        lk_api_key = (
            api_key
            if api_key
            else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
        )
        if not lk_api_key:
            raise ValueError(
                "api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
            )

        lk_api_secret = (
            api_secret
            if api_secret
            else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
        )
        if not lk_api_secret:
            raise ValueError(
                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
            )

        self._opts = InterruptionOptions(
            sample_rate=SAMPLE_RATE,
            threshold=threshold,
            min_frames=math.ceil(min_interruption_duration * _FRAMES_PER_SECOND),
            max_audio_duration=max_audio_duration,
            audio_prefix_duration=audio_prefix_duration,
            detection_interval=detection_interval,
            inference_timeout=inference_timeout,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
        )
        self._label = f"{type(self).__module__}.{type(self).__name__}"
        self._sample_rate = SAMPLE_RATE
        self._session = http_session
        self._streams = weakref.WeakSet[InterruptionWebSocketStream]()

        logger.info(
            "adaptive interruption detector initialized",
            extra={
                "base_url": self._opts.base_url,
                "detection_interval": self._opts.detection_interval,
                "audio_prefix_duration": self._opts.audio_prefix_duration,
                "max_audio_duration": self._opts.max_audio_duration,
                "min_frames": self._opts.min_frames,
                "threshold": self._opts.threshold if is_given(self._opts.threshold) else None,
                "inference_timeout": self._opts.inference_timeout,
            },
        )

    @property
    def model(self) -> str:
        return "adaptive interruption"

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

    @property
    def label(self) -> str:
        return self._label

    @property
    def sample_rate(self) -> int:
        return self._sample_rate

    def _emit_error(self, api_error: Exception, recoverable: bool) -> None:
        self.emit(
            "error",
            InterruptionDetectionError(
                label=self._label,
                error=api_error,
                recoverable=recoverable,
            ),
        )

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = http_context.http_session()
        return self._session

    def stream(
        self, *, conn_options: APIConnectOptions = DEFAULT_INTERRUPTION_CONN_OPTIONS
    ) -> InterruptionWebSocketStream:
        try:
            stream = InterruptionWebSocketStream(model=self, conn_options=conn_options)
        except Exception as e:
            self._emit_error(e, recoverable=False)
            raise
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        threshold: NotGivenOr[float] = NOT_GIVEN,
        min_interruption_duration: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        if is_given(threshold):
            self._opts.threshold = threshold
        if is_given(min_interruption_duration):
            self._opts.min_frames = math.ceil(min_interruption_duration * _FRAMES_PER_SECOND)

        for stream in self._streams:
            stream.update_options(
                threshold=threshold, min_interruption_duration=min_interruption_duration
            )

class AvatarSession(BaseAvatarSession):
    """An avatar session provisioned through LiveKit Inference.

    Unlike the BYOK avatar plugins (which call the avatar provider directly with
    a customer API key), this calls the LiveKit Inference gateway with LiveKit
    credentials; the gateway creates the provider session with LiveKit's
    wholesale key. Media and RPC still flow in-room over DataStream, exactly as
    the BYOK plugins do.

    Example::

        avatar = inference.AvatarSession(
            "lemonslice",
            extra_kwargs=LemonSliceOptions(image_url="https://...", prompt="..."),
        )
        await avatar.start(session, room=ctx.room)
        await avatar.wait_for_join()

    Unlike the BYOK plugins, this does **not** mint the avatar worker's room
    token. The gateway mints it, scoped to this room and identity and stamped
    with ``lk.publish_on_behalf`` and ``lk.avatar_provider``; a token the client
    signed could not be trusted to carry those. The room fields this class sends
    (``room_name`` / ``avatar_identity`` / ``agent_identity``) are the inputs to
    that mint, so all three are required.

    Credentials are gateway auth only (``api_key`` / ``api_secret``): resolved
    from the arguments, then ``LIVEKIT_INFERENCE_API_KEY`` /
    ``LIVEKIT_INFERENCE_API_SECRET``, then ``LIVEKIT_API_KEY`` /
    ``LIVEKIT_API_SECRET``. Because the gateway signs the worker token with the
    key it authenticated, the project that serves inference must be the project
    that owns the room — pointing ``LIVEKIT_INFERENCE_*`` at a different project
    than the room is no longer supported.
    """

    def __init__(
        self,
        model: str,
        *,
        avatar_participant_identity: NotGivenOr[str] = NOT_GIVEN,
        avatar_participant_name: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[LemonSliceOptions | dict[str, Any]] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            model: ``"<provider>"`` or ``"<provider>/<avatar_id>"``
                (e.g. ``"lemonslice"`` or ``"lemonslice/agent_abc"``). For
                LemonSlice the ``<avatar_id>`` is the agent id (the gateway maps
                it onto LemonSlice's ``agent_id``); pass a pre-built agent as
                ``"lemonslice/<agent_id>"``.
            avatar_participant_identity: Room identity for the avatar worker.
                Defaults to ``"<provider>-inference-avatar"`` (the ``inference``
                marker distinguishes it from the BYOK plugins' worker identity).
            avatar_participant_name: Display name for the avatar worker.
            extra_kwargs: Provider-specific options. For LemonSlice use
                :class:`LemonSliceOptions` (``image_url`` / ``prompt`` /
                ``idle_prompt`` / ``idle_timeout``); known keys map onto the
                gateway's first-class request fields and anything else is
                forwarded verbatim, subject to the gateway's per-provider
                allowlist. ``image_url`` is mutually exclusive with a catalog id
                in the model string — passing both raises here (and the gateway
                rejects the combination too).
            base_url: Inference gateway base URL. Defaults to the environment's
                gateway (see ``get_default_inference_url``).
            api_key: Gateway API key (see the class docstring for resolution).
            api_secret: Gateway API secret (see the class docstring).
            http_session: Optional aiohttp session; one is created per call if
                omitted.
            conn_options: Retry/timeout options for the create call.
        """
        super().__init__()

        self._provider, self._avatar_id = _parse_avatar_model(model)
        # Copy so a later mutation of the caller's dict can't change what we send.
        self._extra_kwargs: dict[str, Any] = dict(extra_kwargs) if is_given(extra_kwargs) else {}
        if self._avatar_id and "image_url" in self._extra_kwargs:
            # The gateway (e.g. the LemonSlice adapter) rejects both; fail fast
            # here so the caller doesn't spend a round trip to learn that.
            raise ValueError(
                "pass either a catalog id in the model string "
                f"('{self._provider}/<avatar_id>') or image_url, not both"
            )
        self._conn_options = conn_options
        self._http_session = http_session

        self._base_url = base_url if is_given(base_url) else get_default_inference_url()
        self._api_key = (
            api_key
            if is_given(api_key)
            else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
        )
        if not self._api_key:
            raise ValueError(
                "api_key is required, either as argument or set LIVEKIT_API_KEY environment variable"
            )
        self._api_secret = (
            api_secret
            if is_given(api_secret)
            else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
        )
        if not self._api_secret:
            raise ValueError(
                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environment variable"
            )

        self._avatar_participant_identity = (
            avatar_participant_identity
            if is_given(avatar_participant_identity)
            else f"{self._provider}-inference-avatar"
        )
        self._avatar_participant_name = (
            avatar_participant_name
            if is_given(avatar_participant_name)
            else self._avatar_participant_identity
        )

        self._session_id: str | None = None
        self._provider_session_id: str | None = None
        # HMAC issued by the gateway alongside provider_session_id; required by
        # /avatar/sessions/terminate to prove this project owns the session.
        self._terminate_token: str | None = None

    @property
    def avatar_identity(self) -> str:
        return self._avatar_participant_identity

    @property
    def provider(self) -> str:
        return self._provider

    @property
    def session_id(self) -> str | None:
        """The gateway-generated avatar session id (available after start())."""
        return self._session_id

    @property
    def provider_session_id(self) -> str | None:
        """The provider's own session id (available after start())."""
        return self._provider_session_id

    async def start(
        self,
        agent_session: AgentSession,
        room: rtc.Room,
        *,
        livekit_url: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        if self._session_id is not None or self._provider_session_id is not None:
            raise RuntimeError(
                "AvatarSession.start() may only be called once per instance; "
                "create a new AvatarSession to start another avatar"
            )

        await super().start(agent_session, room)

        # Only the URL is needed: the provider dials it with the token the
        # gateway mints, so no room-project key/secret is resolved here.
        lk_url = livekit_url or (os.getenv("LIVEKIT_URL") or NOT_GIVEN)
        if not lk_url:
            raise ValueError(
                "livekit_url must be set by argument or the LIVEKIT_URL environment variable"
            )

        # Usable inside an agent job or standalone (scripts/tests). The base
        # class already tolerates a missing job context; derive the same values
        # from the connected room when there is none.
        job_ctx = get_job_context(required=False)
        if job_ctx is not None:
            local_participant_identity = job_ctx.local_participant_identity
            room_sid = job_ctx.job.room.sid or await room.sid
        elif room.isconnected():
            local_participant_identity = room.local_participant.identity
            room_sid = await room.sid
        else:
            raise RuntimeError(
                "AvatarSession.start() needs a connected room or an agent job context; "
                "connect the room before calling start()"
            )

        create_resp = await self._create_session(
            room_name=room.name,
            room_sid=room_sid,
            livekit_url=lk_url,
            agent_identity=local_participant_identity,
        )

        self._session_id = create_resp.get("session_id")
        self._provider_session_id = create_resp.get("provider_session_id")
        self._terminate_token = create_resp.get("terminate_token")
        sample_rate = create_resp.get("sample_rate") or _DEFAULT_SAMPLE_RATE

        if not self._provider_session_id:
            # Without it aclose() has nothing to terminate; the provider session
            # runs to its own idle timeout instead of being stopped explicitly.
            logger.warning(
                "avatar gateway create response had no provider_session_id; this "
                "session cannot be explicitly terminated and will bill until its "
                "provider idle timeout",
                extra={"provider": self._provider, "session_id": self._session_id},
            )
        elif not self._terminate_token:
            # aclose() cannot terminate without this; the provider session will
            # run to its own idle timeout instead of being stopped explicitly.
            logger.warning(
                "avatar gateway create response had no terminate_token; this "
                "session cannot be explicitly terminated and will bill until "
                "its provider idle timeout",
                extra={
                    "provider": self._provider,
                    "session_id": self._session_id,
                    "provider_session_id": self._provider_session_id,
                },
            )

        # Rebind the audio tail to the avatar worker using the gateway-reported
        # sample rate. Done after the create response (the lemonslice BYOK plugin
        # instead hardcodes the rate and rebinds first): in the canonical flow
        # avatar.start() runs before session.start(), so no audio has flowed yet
        # and nothing is lost. wait_remote_track buffers until the video track
        # appears; replace_audio_tail keeps the TranscriptSynchronizer /
        # RecorderAudioOutput chain intact.
        agent_session.output.replace_audio_tail(
            DataStreamAudioOutput(
                room=room,
                destination_identity=self._avatar_participant_identity,
                sample_rate=sample_rate,
                wait_remote_track=rtc.TrackKind.KIND_VIDEO,
                clear_buffer_timeout=None,
                wait_playback_start=True,
            ),
        )

        logger.debug(
            "inference avatar session created",
            extra={
                "provider": self._provider,
                "session_id": self._session_id,
                "provider_session_id": self._provider_session_id,
            },
        )

    async def aclose(self) -> None:
        # Terminate the provider session through the gateway so it stops billing
        # immediately instead of lingering until its idle timeout. Best-effort:
        # a failure must not block participant cleanup in the base class, so
        # base cleanup runs in `finally` even if terminate raises or is
        # cancelled (e.g. job-shutdown deadline). The ids are cleared only on
        # a confirmed terminate so a second aclose() call can still retry.
        provider_session_id = self._provider_session_id
        terminate_token = self._terminate_token
        try:
            if provider_session_id and terminate_token:
                try:
                    await self._terminate_session(provider_session_id, terminate_token)
                except Exception:
                    logger.warning(
                        "failed to terminate inference avatar session; it will "
                        "keep billing until its provider idle timeout unless "
                        "aclose() is called again",
                        extra={
                            "provider": self._provider,
                            "provider_session_id": provider_session_id,
                        },
                        exc_info=True,
                    )
                else:
                    self._provider_session_id = None
                    self._terminate_token = None
            elif provider_session_id:
                logger.debug(
                    "no terminate_token for this avatar session; skipping explicit terminate",
                    extra={
                        "provider": self._provider,
                        "provider_session_id": provider_session_id,
                    },
                )
        finally:
            await super().aclose()

    def _auth_headers(self) -> dict[str, str]:
        # Minted fresh on each call: callers inside the create retry loop re-mint
        # per attempt rather than reusing a token that may expire between tries.
        return {
            **get_inference_headers(),
            "Authorization": f"Bearer {create_access_token(self._api_key, self._api_secret)}",
            HEADER_INFERENCE_PROVIDER: self._provider,
            "Content-Type": "application/json",
        }

    async def _create_session(
        self,
        *,
        room_name: str,
        room_sid: str,
        livekit_url: str,
        agent_identity: str,
    ) -> dict[str, Any]:
        # room_name / avatar_identity / agent_identity are the inputs the
        # gateway mints the worker room token from, not just attribution: it
        # scopes the roomJoin grant to room_name, joins as avatar_identity, and
        # sets lk.publish_on_behalf to agent_identity.
        payload: dict[str, Any] = {
            "provider": self._provider,
            "livekit_url": livekit_url,
            "room_name": room_name,
            "room_sid": room_sid,
            "avatar_identity": self._avatar_participant_identity,
            "avatar_name": self._avatar_participant_name,
            "agent_identity": agent_identity,
        }
        if self._avatar_id:
            payload["avatar_id"] = self._avatar_id
        # Map known LemonSlice options onto the gateway's first-class request
        # fields; forward anything else via extra_kwargs (gateway allowlist).
        extra: dict[str, Any] = {}
        for key, value in self._extra_kwargs.items():
            payload_field = _OPTION_TO_PAYLOAD_FIELD.get(key)
            if payload_field is not None:
                payload[payload_field] = value
            else:
                extra[key] = value
        if extra:
            payload["extra_kwargs"] = extra

        # One idempotency key per start(), stable across retries, so a retried
        # create replays the first result on the gateway instead of paying for a
        # second provider session.
        idempotency_key = uuid.uuid4().hex
        url = f"{self._base_url.rstrip('/')}/avatar/sessions"

        session = self._http_session or aiohttp.ClientSession()
        try:
            last_exc: Exception | None = None
            for i in range(self._conn_options.max_retry + 1):
                headers = {**self._auth_headers(), "Idempotency-Key": idempotency_key}
                try:
                    async with session.post(
                        url,
                        headers=headers,
                        json=payload,
                        timeout=aiohttp.ClientTimeout(
                            total=_REQUEST_TIMEOUT,
                            sock_connect=self._conn_options.timeout,
                        ),
                    ) as response:
                        if not response.ok:
                            text = await response.text()
                            raise create_api_error_from_http(
                                f"avatar gateway returned an error: {text}",
                                status=response.status,
                                body=text,
                            )
                        return await response.json()  # type: ignore[no-any-return]
                except asyncio.TimeoutError:
                    last_exc = APITimeoutError(
                        f"avatar gateway create timed out after attempt {i + 1}"
                    )
                    logger.warning(
                        "avatar gateway request timed out",
                        extra={"provider": self._provider, "attempt": i},
                    )
                except aiohttp.ClientError as e:
                    last_exc = APIConnectionError(str(e))
                    logger.warning(
                        "failed to call avatar gateway",
                        extra={"provider": self._provider, "error": str(e)},
                    )
                except APIError as e:
                    last_exc = e
                    if not e.retryable:
                        raise
                    logger.warning(
                        "avatar gateway returned a retryable error",
                        extra={"provider": self._provider, "error": str(e)},
                    )

                if i < self._conn_options.max_retry:
                    await asyncio.sleep(self._conn_options._interval_for_retry(i))

            if last_exc is not None:
                raise last_exc
            raise APIConnectionError("failed to create avatar session after all retries")
        finally:
            if self._http_session is None:
                await session.close()

    async def _terminate_session(self, provider_session_id: str, terminate_token: str) -> None:
        url = f"{self._base_url.rstrip('/')}/avatar/sessions/terminate"
        headers = self._auth_headers()
        payload = {
            "provider": self._provider,
            "provider_session_id": provider_session_id,
            "terminate_token": terminate_token,
        }

        session = self._http_session or aiohttp.ClientSession()
        try:
            async with session.post(
                url,
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(
                    total=_REQUEST_TIMEOUT, sock_connect=self._conn_options.timeout
                ),
            ) as response:
                if not response.ok:
                    text = await response.text()
                    raise create_api_error_from_http(
                        f"avatar gateway returned an error: {text}",
                        status=response.status,
                        body=text,
                    )
        finally:
            if self._http_session is None:
                await session.close()

class DecisionModel(BaseDecisionModel):
    """Decision models served through LiveKit Inference (``POST /v1/decisions``).

    Unlike the BYOK plugins, this calls the inference gateway with LiveKit
    credentials; the gateway calls the provider with LiveKit's key. Every provider
    speaks the same body, so ``model`` is the only thing that changes between them.

    Example::

        model = inference.DecisionModel("openai/gpt-6-luna-decisions")
        response = await model.evaluate(
            input=chat_ctx,
            decisions={"handoff": llm.Probability("The caller requests a human.")},
        )

    Credentials resolve from the arguments, then ``LIVEKIT_INFERENCE_API_KEY`` /
    ``LIVEKIT_INFERENCE_API_SECRET``, then ``LIVEKIT_API_KEY`` / ``LIVEKIT_API_SECRET``.
    A project the endpoint is not enabled for gets a non-retryable 403.
    """

    def __init__(
        self,
        model: DecisionModels | str = "openai/gpt-6-luna-decisions",
        *,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
    ) -> None:
        """
        Args:
            model: Gateway model id, ``"<provider>/<model>"``.
            base_url: Inference gateway base URL. Defaults to the environment's
                gateway (see ``get_default_inference_url``).
            api_key: Gateway API key (see the class docstring for resolution).
            api_secret: Gateway API secret (see the class docstring).
            http_session: Optional caller-owned session. Otherwise uses the job's
                HTTP session. Outside a job, use ``async with utils.http_context.open()``.
        """
        super().__init__(
            capabilities=DecisionCapabilities(decision_kinds=["probability", "choice", "score"])
        )
        self._model = model
        self._base_url = base_url if is_given(base_url) else get_default_inference_url()
        self._api_key, self._api_secret = resolve_credentials(
            api_key if is_given(api_key) else None,
            api_secret if is_given(api_secret) else None,
        )
        self._session = http_session

    @property
    def model(self) -> str:
        return self._model

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

    def _headers(self) -> dict[str, str]:
        # Minted per request: a retry must not reuse a token that may have expired.
        return {
            **get_inference_headers(),
            "Authorization": f"Bearer {create_access_token(self._api_key, self._api_secret)}",
            "Content-Type": "application/json",
        }

    async def _evaluate_impl(
        self,
        *,
        input: str,
        decisions: Mapping[str, Decision],
        conn_options: APIConnectOptions,
    ) -> DecisionResponse:
        questions = {name: systemone.encode_question(d) for name, d in decisions.items()}
        session = self._session or http_context.http_session()
        url = f"{self._base_url.rstrip('/')}/decisions"
        try:
            async with session.post(
                url,
                headers=self._headers(),
                json={"model": self._model, "state": input, "questions": questions},
                timeout=aiohttp.ClientTimeout(total=conn_options.timeout),
            ) as response:
                request_id = response.headers.get(HEADER_REQUEST_ID)
                if response.status >= 400:
                    raise _error_from_response(
                        await response.text(), status=response.status, request_id=request_id
                    )
                parsed = systemone.Response.model_validate(await response.json())
            results, errors = systemone.decode_answers(parsed.answers)
            return DecisionResponse(
                results=results,
                errors=errors,
                model=parsed.model,
                provider=self.provider,
                request_id=request_id or parsed.id,
                input_tokens=parsed.usage.input_tokens,
                output_tokens=parsed.usage.output_tokens,
            )
        except asyncio.TimeoutError as exc:
            raise APITimeoutError("inference decision request timed out") from exc
        except aiohttp.ContentTypeError as exc:
            raise APIError("invalid inference decision response", retryable=False) from exc
        except aiohttp.ClientError as exc:
            raise APIConnectionError("inference gateway connection failed") from exc
        except ValueError as exc:
            raise APIError("invalid inference decision response", retryable=False) from exc

class DuplexModel(_GPTLiveModel):
    """Full-duplex voice model, served through LiveKit Inference."""

    @overload
    def __init__(
        self,
        model: OpenAIDuplexModels,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[OpenAIDuplexOptions] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        pass

    def __init__(
        self,
        model: DuplexModels | str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[OpenAIDuplexOptions | dict[str, Any]] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            model: Provider-prefixed duplex model, for example ``openai/gpt-live-1``.
            provider: Host that must serve the model, e.g. ``"openai"``. The session is
                refused if no host in the region matches.
            voice: Output voice, ``marin`` by default for GPT-Live.
            extra_kwargs: Model-specific options; :class:`OpenAIDuplexOptions` for GPT-Live.
                Delegated work runs on the default service tier only.
            http_session: Optional shared HTTP session.
            max_session_duration: Seconds before the connection is recycled.
            conn_options: Retry/backoff and connection settings.
        """
        if "/" not in model:
            raise ValueError("model must be provider-prefixed, for example 'openai/gpt-live-1'")
        options: dict[str, Any] = dict(extra_kwargs) if is_given(extra_kwargs) else {}
        if unknown := options.keys() - OpenAIDuplexOptions.__optional_keys__:
            raise ValueError(f"unsupported extra_kwargs for {model}: {sorted(unknown)}")
        responses_options = options.get("responses_options")
        if responses_options and responses_options.get("service_tier") not in (None, "default"):
            raise ValueError("LiveKit Inference runs GPT-Live delegation on the default tier only")

        resolved_api_key, resolved_api_secret = resolve_credentials(api_key, api_secret)

        super().__init__(
            model=model,
            voice=voice if is_given(voice) else DEFAULT_VOICE,
            delegation=options.get("delegation", "responses"),
            responses_options=responses_options if responses_options is not None else NOT_GIVEN,
            api_key="livekit-inference",
            base_url=base_url or get_default_inference_url(),
            http_session=http_session,
            max_session_duration=max_session_duration,
            conn_options=conn_options,
        )
        self._inference_opts = _InferenceOptions(
            provider=provider,
            api_key=resolved_api_key,
            api_secret=resolved_api_secret,
        )
        self._provider_label = "LiveKit Inference Duplex"

    @classmethod
    def from_model_string(cls, model: str) -> DuplexModel:
        """Create a DuplexModel instance from a model string"""
        return cls(model)

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

    def session(self) -> DuplexSession:
        return DuplexSession(self)

class DuplexSession(_GPTLiveSession):
    def __init__(self, duplex_model: DuplexModel) -> None:
        self._inference_model = duplex_model
        super().__init__(duplex_model)

    def _create_ws_url_and_headers(self) -> tuple[str, dict[str, str]]:
        opts = self._inference_model._inference_opts
        url, _ = super()._create_ws_url_and_headers()
        url = f"{url}?{urlencode({'model': self._opts.model})}"
        headers = get_inference_headers()
        # signed per connection, so a reconnect never presents an expired token
        headers["Authorization"] = f"Bearer {create_access_token(opts.api_key, opts.api_secret)}"
        if opts.provider:
            headers[HEADER_INFERENCE_PROVIDER] = opts.provider
        return url, headers

    async def _update_tools(self, tools: list[llm.Tool]) -> None:
        hosted = sorted(
            tool.id
            for tool in llm.ToolContext(tools).flatten()
            if isinstance(tool, llm.ProviderTool)
        )
        if hosted:
            # hosted tools bill separately from the delegated tokens the gateway meters
            raise llm.RealtimeError(
                f"LiveKit Inference does not run hosted tools on GPT-Live, so the model can never "
                f"call {hosted}. Use function tools, or the OpenAI plugin's GPTLiveModel directly."
            )
        await super()._update_tools(tools)

    def _is_fatal_error(self, error: types.ErrorBody) -> bool:
        return (error.code or "") in _FATAL_ERROR_CODES or super()._is_fatal_error(error)

class InterruptionDetectionError(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["interruption_detection_error"] = "interruption_detection_error"
    timestamp: float = Field(default_factory=time.time)
    label: str
    error: Exception = Field(..., exclude=True)
    recoverable: bool

class LLM(llm.LLM):
    def __init__(
        self,
        model: LLMModels | str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        inference_class: InferenceClass | None = None,
        extra_kwargs: ChatCompletionOptions | dict[str, Any] | None = None,
        prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
    ) -> None:
        """
        Args:
            prompt_cache_breakpoints: Send each :class:`llm.CacheBreakpoint` in the chat
                context as an OpenAI ``prompt_cache_breakpoint``. ``"auto"`` sends them
                when :func:`supports_prompt_cache_breakpoints` accepts the current model;
                ``True`` or ``False`` overrides that.
        """
        super().__init__()

        lk_base_url = base_url if base_url else get_default_inference_url()

        lk_api_key, lk_api_secret = resolve_credentials(api_key, api_secret)

        self._opts = _LLMOptions(
            model=model,
            provider=provider,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            inference_class=inference_class,
            extra_kwargs=extra_kwargs or {},
            prompt_cache_breakpoints=prompt_cache_breakpoints,
        )
        self._client = openai.AsyncClient(
            api_key=create_access_token(self._opts.api_key, self._opts.api_secret),
            base_url=self._opts.base_url,
            max_retries=0,
            http_client=httpx.AsyncClient(
                timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
                follow_redirects=True,
                limits=httpx.Limits(
                    max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
                ),
            ),
        )

    async def _prewarm_impl(self) -> None:
        await self._client.models.list()

    async def aclose(self) -> None:
        await super().aclose()
        await self._client.close()

    @classmethod
    def from_model_string(cls, model: str) -> LLM:
        """Create a LLM instance from a model string"""
        return cls(model)

    def update_options(
        self,
        *,
        model: NotGivenOr[LLMModels | str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[ChatCompletionOptions | dict[str, Any]] = NOT_GIVEN,
        prompt_cache_breakpoints: NotGivenOr[bool | Literal["auto"]] = NOT_GIVEN,
    ) -> None:
        """Update LLM configuration options.

        Each option is read on the next ``chat()`` call, so a swap
        takes effect on the agent's next turn without recreating the
        LLM. ``extra_kwargs`` *replaces* the persistent kwargs dict
        rather than merging — pass ``{}`` to clear it.
        """
        if is_given(model):
            self._opts.model = model
        if is_given(extra_kwargs):
            self._opts.extra_kwargs = dict(extra_kwargs)
        # isinstance rather than is_given: mypy cannot narrow the bool | Literal union through it
        if not isinstance(prompt_cache_breakpoints, NotGiven):
            self._opts.prompt_cache_breakpoints = prompt_cache_breakpoints

    @property
    def model(self) -> str:
        """Get the model name for this LLM instance."""
        return self._opts.model

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

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        response_format: NotGivenOr[
            completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]
        ] = NOT_GIVEN,
        inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        extra = {}
        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        parallel_tool_calls = (
            parallel_tool_calls
            if is_given(parallel_tool_calls)
            else self._opts.extra_kwargs.get("parallel_tool_calls", NOT_GIVEN)
        )
        if is_given(parallel_tool_calls):
            extra["parallel_tool_calls"] = parallel_tool_calls

        extra_tool_choice = self._opts.extra_kwargs.get("tool_choice", NOT_GIVEN)
        tool_choice = tool_choice if is_given(tool_choice) else extra_tool_choice
        if is_given(tool_choice):
            oai_tool_choice: ChatCompletionToolChoiceOptionParam
            if isinstance(tool_choice, dict):
                oai_tool_choice = {
                    "type": "function",
                    "function": {"name": tool_choice["function"]["name"]},
                }
                extra["tool_choice"] = oai_tool_choice
            elif tool_choice in ("auto", "required", "none"):
                oai_tool_choice = tool_choice
                extra["tool_choice"] = oai_tool_choice

        if is_given(response_format):
            extra["response_format"] = llm_utils.to_openai_response_format(response_format)  # type: ignore

        extra.update(self._opts.extra_kwargs)

        effective_inference_class = (
            inference_class if is_given(inference_class) else self._opts.inference_class
        )

        self._client.api_key = create_access_token(self._opts.api_key, self._opts.api_secret)
        return LLMStream(
            self,
            model=self._opts.model,
            provider=self._opts.provider,
            inference_class=effective_inference_class,
            strict_tool_schema=True,
            client=self._client,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            extra_kwargs=extra,
            prompt_cache_breakpoints=self._resolve_prompt_cache_breakpoints(),
        )

    def _resolve_prompt_cache_breakpoints(self) -> bool:
        # resolved per chat() so update_options(model=...) re-evaluates "auto"
        setting = self._opts.prompt_cache_breakpoints
        if isinstance(setting, bool):
            return setting
        return supports_prompt_cache_breakpoints(self._opts.model)

class LLMStream(llm.LLMStream):
    def __init__(
        self,
        llm_v: LLM | llm.LLM,
        *,
        model: LLMModels | str,
        provider: str | None = None,
        inference_class: InferenceClass | None = None,
        strict_tool_schema: bool,
        client: openai.AsyncClient,
        chat_ctx: llm.ChatContext,
        tools: list[Tool],
        conn_options: APIConnectOptions,
        extra_kwargs: dict[str, Any],
        provider_fmt: str = "openai",  # used internally for chat_ctx format
        prompt_cache_breakpoints: bool = False,
    ) -> None:
        super().__init__(llm_v, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
        self._model = model
        self._provider = provider
        self._inference_class = inference_class
        self._provider_fmt = provider_fmt
        # the other formatters take no such argument and drop CacheBreakpoint themselves
        self._prompt_cache_breakpoints = prompt_cache_breakpoints and (
            provider_fmt in _BREAKPOINT_FORMATS
        )
        self._strict_tool_schema = strict_tool_schema
        self._client = client
        self._llm = llm_v
        self._extra_kwargs = drop_unsupported_params(model, extra_kwargs, tools=tools)
        self._tool_ctx = llm.ToolContext(tools)

    async def _run(self) -> None:
        # current function call that we're waiting for full completion (args are streamed)
        # (defined inside the _run method to make sure the state is reset for each run/attempt)
        self._oai_stream: openai.AsyncStream[ChatCompletionChunk] | None = None
        self._tool_call_id: str | None = None
        self._fnc_name: str | None = None
        self._fnc_raw_arguments: str | None = None
        self._tool_extra: dict[str, Any] | None = None
        self._tool_index: int | None = None
        retryable = True

        try:
            fmt_kwargs: dict[str, Any] = {}
            if self._prompt_cache_breakpoints:
                fmt_kwargs["prompt_cache_breakpoints"] = True
            chat_ctx, _ = self._chat_ctx.to_provider_format(format=self._provider_fmt, **fmt_kwargs)
            tool_schemas = cast(
                list[ChatCompletionToolParam],
                self._tool_ctx.parse_function_tools("openai", strict=self._strict_tool_schema),
            )
            if lk_oai_debug:
                tool_choice = self._extra_kwargs.get("tool_choice", NOT_GIVEN)
                logger.debug(
                    "chat.completions.create",
                    extra={
                        "fnc_ctx": tool_schemas,
                        "tool_choice": tool_choice,
                        "lk.pii.chat_ctx": chat_ctx,
                    },
                )
            if not self._tools:
                # remove tool_choice from extra_kwargs if no tools are provided
                self._extra_kwargs.pop("tool_choice", None)

            extra_headers = self._extra_kwargs.setdefault("extra_headers", {})
            extra_headers.update(get_inference_headers(inference_class=self._inference_class))
            if self._provider:
                extra_headers[HEADER_INFERENCE_PROVIDER] = self._provider

            self._oai_stream = stream = await self._client.chat.completions.create(
                messages=cast(list[ChatCompletionMessageParam], chat_ctx),
                tools=tool_schemas or openai.omit,
                model=self._model,
                stream_options={"include_usage": True},
                stream=True,
                timeout=httpx.Timeout(self._conn_options.timeout),
                **self._extra_kwargs,
            )

            thinking_filter = llm_utils.ThinkingTokenFilter(
                *_MODEL_THINK_TAGS.get(
                    self._model, (llm_utils.THINK_TAG_START, llm_utils.THINK_TAG_END)
                )
            )
            async with stream:
                async for chunk in stream:
                    for choice in chunk.choices:
                        chat_chunk = self._parse_choice(chunk.id, choice, thinking_filter)
                        if chat_chunk is not None:
                            if chat_chunk.has_response():
                                retryable = False
                            self._event_ch.send_nowait(chat_chunk)

                    if chunk.usage is not None:
                        tokens_details = chunk.usage.prompt_tokens_details
                        cached_tokens = tokens_details.cached_tokens if tokens_details else 0
                        cache_creation_tokens = (
                            tokens_details.cache_write_tokens if tokens_details else 0
                        )
                        completion_details = chunk.usage.completion_tokens_details
                        reasoning_tokens = (
                            completion_details.reasoning_tokens if completion_details else 0
                        )
                        usage_chunk = llm.ChatChunk(
                            id=chunk.id,
                            usage=llm.CompletionUsage(
                                completion_tokens=chunk.usage.completion_tokens or 0,
                                prompt_tokens=chunk.usage.prompt_tokens or 0,
                                prompt_cached_tokens=cached_tokens or 0,
                                cache_creation_tokens=cache_creation_tokens or 0,
                                reasoning_tokens=reasoning_tokens or 0,
                                total_tokens=chunk.usage.total_tokens or 0,
                                service_tier=getattr(chunk, "service_tier", None),
                            ),
                        )
                        self._event_ch.send_nowait(usage_chunk)

        except openai.APITimeoutError:
            raise APITimeoutError(retryable=retryable) from None
        except httpx.TimeoutException as e:
            # Only the request call runs inside the openai client's error mapping, so a
            # timeout waiting on the stream body arrives as the raw httpx exception.
            raise APITimeoutError(retryable=retryable) from e
        except openai.APIStatusError as e:
            if e.status_code == 429:
                self._log_rate_limited(e)
            raise APIStatusError(
                e.message,
                status_code=e.status_code,
                request_id=e.request_id,
                body=e.body,
                retryable=retryable,
            ) from None
        except Exception as e:
            raise APIConnectionError(retryable=retryable) from e

    def _log_rate_limited(self, e: openai.APIStatusError) -> None:
        """Log the gateway's quota snapshot when a request is rejected with 429.

        The gateway stamps X-LiveKit-Inference-{RPM,TPM,Credits}-{Limit,Used}
        on rejections so customers can see which limit they hit and by how much.
        """
        extra: dict[str, Any] = {"model": self._model, "request_id": e.request_id}
        extra.update(extract_quota_usage(e.response.headers))
        logger.warning("LLM request rate limited by inference gateway", extra=extra)

    def _parse_choice(
        self, id: str, choice: Choice, thinking_filter: llm_utils.ThinkingTokenFilter
    ) -> llm.ChatChunk | None:
        delta = choice.delta

        # https://github.com/livekit/agents/issues/688
        # the delta can be None when using Azure OpenAI (content filtering)
        if delta is None:
            return None

        delta.content = llm_utils.strip_thinking_tokens(
            delta.content, thinking_filter, final=choice.finish_reason is not None
        )

        # Calls completed by this delta. A provider can send several parallel calls
        # in one chunk (e.g. Ollama), so they are collected and emitted together.
        completed: list[llm.FunctionToolCall] = []

        if delta.tool_calls:
            for tool in delta.tool_calls:
                if not tool.function:
                    continue

                if self._tool_call_id and tool.id and tool.index != self._tool_index:
                    completed.append(
                        llm.FunctionToolCall(
                            arguments=self._fnc_raw_arguments or "",
                            name=self._fnc_name or "",
                            call_id=self._tool_call_id or "",
                            extra=self._tool_extra,
                        )
                    )
                    self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
                    self._tool_extra = None

                if tool.function.name:
                    self._tool_index = tool.index
                    self._tool_call_id = tool.id
                    self._fnc_name = tool.function.name
                    self._fnc_raw_arguments = tool.function.arguments or ""
                    # Extract extra from tool call (e.g., Google thought signatures)
                    self._tool_extra = getattr(tool, "extra_content", None)
                elif tool.function.arguments:
                    self._fnc_raw_arguments += tool.function.arguments  # type: ignore

        if choice.finish_reason in ("tool_calls", "stop") and self._tool_call_id:
            completed.append(
                llm.FunctionToolCall(
                    arguments=self._fnc_raw_arguments or "",
                    name=self._fnc_name or "",
                    call_id=self._tool_call_id or "",
                    extra=self._tool_extra,
                )
            )
            self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
            self._tool_extra = None

        if completed:
            return llm.ChatChunk(
                id=id,
                delta=llm.ChoiceDelta(
                    role="assistant",
                    content=delta.content,
                    extra=getattr(delta, "extra_content", None),
                    tool_calls=completed,
                ),
            )

        # Extract extra from delta (e.g., Google thought signatures on text parts)
        delta_extra = getattr(delta, "extra_content", None)

        if not delta.content and not delta_extra:
            return None

        return llm.ChatChunk(
            id=id,
            delta=llm.ChoiceDelta(
                content=delta.content,
                role="assistant",
                extra=delta_extra,
            ),
        )

class LemonSliceOptions(TypedDict, total=False):
    """Provider-specific options for a LemonSlice inference avatar.

    Mirrors the per-provider option dicts in :mod:`inference.stt` /
    :mod:`inference.tts` (e.g. ``CartesiaOptions``); passed via ``extra_kwargs``.
    All keys are optional.
    """

    image_url: str  # appearance source; mutually exclusive with a model-string agent id
    prompt: str  # speaking prompt (mapped to the provider's agent_prompt)
    idle_prompt: str  # idle prompt (mapped to the provider's agent_idle_prompt)
    idle_timeout: int  # provider idle timeout in seconds; the gateway clamps it

class OverlappingSpeechEvent(BaseModel):
    """Represents an overlapping speech event detected during agent speech."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    type: Literal["overlapping_speech"] = "overlapping_speech"

    created_at: float = Field(default_factory=time.time)
    """Timestamp (in seconds) when the event was emitted."""

    detected_at: float = Field(default_factory=time.time)
    """Timestamp (in seconds) when the overlap was detected."""

    is_interruption: bool = False
    """Whether interruption is detected."""

    agent_ended: bool = False
    """True when the overlap ended because the agent finished speaking rather than the user.

    The user may still be talking, so ``is_interruption`` (always ``False`` here) is inconclusive
    and must not be treated as a confirmed backchannel verdict.
    """

    total_duration: float = 0.0
    """RTT (Round Trip Time) time taken to perform the inference, in seconds."""

    prediction_duration: float = 0.0
    """Time taken to perform the inference from the model side, in seconds."""

    detection_delay: float = 0.0
    """Total time from the onset of the speech to the final prediction, in seconds."""

    overlap_started_at: float | None = None
    """Timestamp (in seconds) when the overlap speech started. Useful for emitting held transcripts."""

    speech_input: npt.NDArray[np.int16] | None = None
    """The audio input that was used for the inference."""

    probabilities: npt.NDArray[np.float32] | None = None
    """The raw probabilities for the interruption detection."""

    probability: float = 0.0
    """The conservative estimated probability of the interruption event."""

    num_requests: int = 0
    """Number of requests sent for this event."""

    @model_serializer(mode="wrap")
    def serialize_model(self, handler: SerializerFunctionWrapHandler) -> Any:
        # remove numpy arrays from the model dump
        copy = self.model_copy(deep=True)
        data = copy.speech_input, copy.probabilities
        copy.speech_input, copy.probabilities = None, None
        try:
            serialized = handler(copy)
        finally:
            copy.speech_input, copy.probabilities = data
        return serialized

    @classmethod
    def from_cache_entry(
        cls,
        *,
        entry: InterruptionCacheEntry,
        is_interruption: bool,
        started_at: float | None = None,
        ended_at: float | None = None,
        agent_ended: bool = False,
    ) -> OverlappingSpeechEvent:
        """Initialize the event from a cache entry.

        Args:
            entry: The cache entry to initialize the event from.
            is_interruption: Whether the interruption is detected.
            started_at: The timestamp when the overlap speech started.
            ended_at: The timestamp when the overlap speech ended.
            agent_ended: Whether the overlap ended because the agent finished speaking.

        Returns:
            The initialized event.
        """
        return cls(
            type="overlapping_speech",
            detected_at=ended_at or time.time(),
            is_interruption=is_interruption,
            agent_ended=agent_ended,
            overlap_started_at=started_at,
            speech_input=entry.speech_input,
            probabilities=entry.probabilities,
            total_duration=entry.get_total_duration(),
            detection_delay=entry.get_detection_delay(),
            prediction_duration=entry.get_prediction_duration(),
            probability=entry.get_probability(),
        )

class RealtimeModel(_RealtimeModel):
    """OpenAI-compatible realtime model authenticated through LiveKit Inference."""

    def __init__(
        self,
        model: RealtimeModels | str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        inference_class: NotGivenOr[InferenceClass | None] = NOT_GIVEN,
        voice: NotGivenOr[str] = NOT_GIVEN,
        modalities: NotGivenOr[list[Literal["text", "audio"]]] = NOT_GIVEN,
        input_audio_transcription: NotGivenOr[AudioTranscription | None] = NOT_GIVEN,
        input_audio_noise_reduction: NotGivenOr[
            NoiseReductionType | NoiseReduction | None
        ] = NOT_GIVEN,
        turn_detection: NotGivenOr[RealtimeAudioInputTurnDetection | None] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN,
        speed: NotGivenOr[float] = NOT_GIVEN,
        tracing: NotGivenOr[Tracing | None] = NOT_GIVEN,
        truncation: NotGivenOr[RealtimeTruncation | None] = NOT_GIVEN,
        reasoning: NotGivenOr[RealtimeReasoning | None] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            provider (str, optional): Host that must serve the model, e.g. ``"azure"`` or
                ``"openai"`` for OpenAI models. The session is refused if no host in the
                region matches.
        """
        if "/" not in model:
            raise ValueError("model must be provider-prefixed, for example 'openai/gpt-realtime'")
        if is_duplex_model(model):
            raise ValueError(f"{model} is a duplex model; use inference.DuplexModel")
        if replacement := DEPRECATED_REALTIME_MODELS.get(model):
            logger.warning(f"{model} is deprecated, use {replacement} instead")
        if is_given(inference_class):
            logger.warning(
                "inference_class is deprecated and will be removed in v2.0. "
                "It has no effect on realtime sessions."
            )

        resolved_api_key, resolved_api_secret = resolve_credentials(api_key, api_secret)

        is_xai = model.startswith("xai/")
        resolved_voice = voice if is_given(voice) else "eve" if is_xai else DEFAULT_VOICE
        resolved_transcription = input_audio_transcription
        resolved_turn_detection = turn_detection
        # Preserve whether xAI's server VAD is a default the framework may disable.
        can_disable_turn_detection = not is_given(turn_detection)
        if is_xai:
            if not is_given(resolved_transcription):
                resolved_transcription = _XAI_DEFAULT_INPUT_AUDIO_TRANSCRIPTION
            if not is_given(resolved_turn_detection):
                resolved_turn_detection = _XAI_DEFAULT_TURN_DETECTION

        super().__init__(
            model=model,
            voice=resolved_voice,
            modalities=modalities,
            input_audio_transcription=resolved_transcription,
            input_audio_noise_reduction=input_audio_noise_reduction,
            turn_detection=resolved_turn_detection,
            tool_choice=tool_choice,
            speed=speed,
            tracing=tracing,
            truncation=truncation,
            reasoning=reasoning,
            api_key="livekit-inference",
            base_url=base_url or get_default_inference_url(),
            http_session=http_session,
            max_session_duration=max_session_duration,
            conn_options=conn_options,
        )
        # LiveKit Inference always uses the OpenAI-compatible protocol; ambient Azure
        # settings must not change its URL or session wire format.
        self._opts.is_azure = False
        self._opts.api_version = None
        if is_xai:
            self._capabilities.can_disable_turn_detection = can_disable_turn_detection
        self._inference_opts = _InferenceOptions(
            provider=provider,
            api_key=resolved_api_key,
            api_secret=resolved_api_secret,
            is_xai=is_xai,
        )
        self._provider_label = "LiveKit Inference Realtime"

    @classmethod
    def from_model_string(cls, model: str) -> RealtimeModel:
        """Create a RealtimeModel instance from a model string"""
        return cls(model)

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

    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
        sess = RealtimeSession(self, turn_detection_disabled=turn_detection_disabled)
        self._sessions.add(sess)
        return sess

class RealtimeSession(_RealtimeSession):
    # the base __init__ resets the input turn state before the subclass body runs
    _is_xai: bool = False
    _pending_transcription: ConversationItemInputAudioTranscriptionCompletedEvent | None = None
    _response_spoke: bool = False

    def __init__(
        self,
        realtime_model: RealtimeModel,
        *,
        turn_detection_disabled: bool = False,
    ) -> None:
        self._inference_model = realtime_model
        self._is_xai = realtime_model._inference_opts.is_xai
        super().__init__(realtime_model, turn_detection_disabled=turn_detection_disabled)

    def _reset_input_turn_state(self) -> None:
        if self._is_xai:
            self._flush_input_transcription()
        super()._reset_input_turn_state()
        self._response_spoke = False

    async def aclose(self) -> None:
        if self._is_xai:
            self._flush_input_transcription()
        await super().aclose()

    def _create_ws_url_and_headers(self) -> tuple[str, dict[str, str]]:
        opts = self._inference_model._inference_opts
        url, _ = super()._create_ws_url_and_headers()
        headers = get_inference_headers()
        headers["Authorization"] = f"Bearer {create_access_token(opts.api_key, opts.api_secret)}"
        if opts.provider:
            headers[HEADER_INFERENCE_PROVIDER] = opts.provider
        return url, headers

    def _wrap_session_update(self, event_id: str, session: Any) -> dict[str, Any]:
        event = super()._wrap_session_update(event_id, session)
        if hasattr(event, "model_dump"):
            event = event.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=False)
        else:
            event = dict(event)

        session_payload = event.get("session")
        if isinstance(session_payload, dict):
            session_payload.pop("model", None)
        return event

    def _create_update_chat_ctx_events(
        self, chat_ctx: llm.ChatContext
    ) -> list[ConversationItemCreateEvent | ConversationItemDeleteEvent]:
        pending = self._pending_transcription if self._is_xai else None
        node = self._remote_chat_ctx.get(pending.item_id) if pending else None
        if node is not None and chat_ctx.get_by_id(node.item.id) is None:
            chat_ctx = chat_ctx.copy()
            index, previous = 0, node._prev
            while previous is not None:
                if (at := chat_ctx.index_by_id(previous.item.id)) is not None:
                    index = at + 1
                    break
                previous = previous._prev
            chat_ctx.items.insert(index, node.item.model_copy())
        return super()._create_update_chat_ctx_events(chat_ctx)

    def _discard_abandoned_response(self) -> None:
        generation = self._current_generation
        if (
            generation is None
            or self._response_spoke
            or isinstance(generation, _DiscardedGeneration)
        ):
            return
        logger.debug("discarding the response xAI left in flight")
        self._close_current_generation()
        self._current_generation = _DiscardedGeneration()

    def interrupt(self) -> None:
        super().interrupt()
        if self._is_xai:
            self._discard_abandoned_response()

    def _handle_response_created(self, event: ResponseCreatedEvent) -> None:
        if self._is_xai:
            self._discard_abandoned_response()
            self._close_current_generation()
            self._response_spoke = False
        super()._handle_response_created(event)

    def _handle_input_audio_buffer_speech_started(
        self, event: InputAudioBufferSpeechStartedEvent
    ) -> None:
        if not self._is_xai:
            super()._handle_input_audio_buffer_speech_started(event)
            return

        if self._pending_transcription and self._pending_transcription.item_id != event.item_id:
            self._flush_input_transcription()

        started_at = self._input_speech_started_at.get(event.item_id)
        super()._handle_input_audio_buffer_speech_started(event)
        if started_at is not None:
            self._input_speech_started_at[event.item_id] = started_at

    def _handle_conversion_item_added(self, event: ConversationItemAdded) -> None:
        if self._is_xai:
            if event.previous_item_id and not self._remote_chat_ctx.get(event.previous_item_id):
                logger.warning(
                    "xAI placed a chat history item after an unknown item; appended it instead",
                    extra={"item_id": event.item.id, "previous_item_id": event.previous_item_id},
                )
                event.previous_item_id = None
            if event.previous_item_id is None:
                event.previous_item_id = self._remote_chat_ctx.tail_id
        super()._handle_conversion_item_added(event)

    def _handle_conversion_item_deleted(self, event: ConversationItemDeletedEvent) -> None:
        if self._is_xai and event.item_id == "" and self._item_delete_future:
            event.item_id = list(self._item_delete_future.keys())[0]
        super()._handle_conversion_item_deleted(event)

    def _handle_conversion_item_input_audio_transcription_completed(
        self, event: ConversationItemInputAudioTranscriptionCompletedEvent
    ) -> None:
        if not self._is_xai:
            super()._handle_conversion_item_input_audio_transcription_completed(event)
            return

        if getattr(event, "status", None) != "in_progress":
            if self._pending_transcription and self._pending_transcription.item_id != event.item_id:
                self._flush_input_transcription()
            self._pending_transcription = event
        self.emit(
            "input_audio_transcription_completed",
            llm.InputTranscriptionCompleted(
                item_id=event.item_id, transcript=event.transcript, is_final=False
            ),
        )

    def _flush_input_transcription(self) -> None:
        if (event := self._pending_transcription) is None:
            return
        self._pending_transcription = None
        if (remote_item := self._remote_chat_ctx.get(event.item_id)) and (
            remote_item.item.type == "message"
        ):
            remote_item.item.content = [
                c for c in remote_item.item.content if not isinstance(c, str)
            ]
        super()._handle_conversion_item_input_audio_transcription_completed(event)

    def _handle_response_audio_delta(self, event: ResponseAudioDeltaEvent) -> None:
        if self._is_xai:
            self._response_spoke = True
            self._flush_input_transcription()
        super()._handle_response_audio_delta(event)

    def _handle_response_text_delta(self, event: ResponseTextDeltaEvent) -> None:
        if self._is_xai:
            self._response_spoke = True
            self._flush_input_transcription()
        super()._handle_response_text_delta(event)

    def _is_fatal_error(self, error: object | None) -> bool:
        fatal = {
            "unsupported_transcription_model",
            "unsupported_audio_transport",
            "unsupported_audio_format",
        }
        code = getattr(error, "code", None)
        type_ = getattr(error, "type", None)
        return (
            (isinstance(code, str) and code in fatal)
            or (isinstance(type_, str) and type_ in fatal)
            or super()._is_fatal_error(error)
        )

class STT(stt.STT):
    @overload
    def __init__(
        self,
        model: CartesiaModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[CartesiaOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: DeepgramModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[DeepgramOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: DeepgramFluxModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[DeepgramFluxOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: AssemblyAIModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[AssemblyaiOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: XaiModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[XaiOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: SpeechmaticsModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[SpeechmaticsOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: InworldModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[InworldOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: GoogleModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[GoogleOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: OpenAIModels,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[OpenAIOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None: ...

    @overload
    def __init__(
        self,
        model: str,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None: ...

    def __init__(
        self,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[STTEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[
            dict[str, Any]
            | CartesiaOptions
            | DeepgramOptions
            | DeepgramFluxOptions
            | AssemblyaiOptions
            | XaiOptions
            | SpeechmaticsOptions
            | InworldOptions
            | GoogleOptions
            | OpenAIOptions
        ] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None:
        """Livekit Cloud Inference STT

        Args:
            model (STTModels | str, optional): STT model to use, in "provider/model[:language]" format.
            language (str, optional): Language of the STT model.
            encoding (STTEncoding, optional): Encoding of the STT model.
            sample_rate (int, optional): Sample rate of the STT model. Defaults to 24000 for
                OpenAI models, which accept no other rate, and to 16000 otherwise.
            base_url (str, optional): LIVEKIT_URL, if not provided, read from environment variable.
            api_key (str, optional): LIVEKIT_API_KEY, if not provided, read from environment variable.
            api_secret (str, optional): LIVEKIT_API_SECRET, if not provided, read from environment variable.
            http_session (aiohttp.ClientSession, optional): HTTP session to use.
            extra_kwargs (dict, optional): Extra kwargs to pass to the STT model.
            fallback (FallbackModelType, optional): Fallback models - either a list of model names,
                a list of FallbackModel instances.
            conn_options (APIConnectOptions, optional): Connection options for request attempts.
            vad (VAD, optional): External Voice Activity Detector. When provided, each audio
                frame is forwarded to the VAD and `session.finalize` is sent to the inference
                gateway on end of speech. Only applicable to models that detect no turns
                themselves: the Speechmatics RT models (enhanced/standard) and OpenAI
                gpt-live-transcribe. They get a default VAD when none is passed.
        """
        # Infer diarization capability from provider-specific extra_kwargs
        # keys (see _DIARIZATION_EXTRA_KEYS). xAI uses "diarize" (same as
        # Deepgram); AssemblyAI uses "speaker_labels".
        diarization_enabled = _diarization_enabled(
            dict(extra_kwargs) if is_given(extra_kwargs) else None
        )

        # Parse language from model string if provided: "provider/model:language"
        if is_given(model) and isinstance(model, str):
            parsed_model, parsed_language = _parse_model_string(model)
            model = parsed_model
            if is_given(parsed_language) and not is_given(language):
                language = parsed_language

        vad = _resolve_vad_for_model(model, vad if is_given(vad) else None)
        resolved_sample_rate = (
            sample_rate
            if is_given(sample_rate)
            else (_required_sample_rate(model) or DEFAULT_SAMPLE_RATE)
        )
        _warn_if_sample_rate_refused(model, resolved_sample_rate)

        fallback_models: NotGivenOr[list[FallbackModel]] = NOT_GIVEN
        if is_given(fallback):
            fallback_models = _normalize_fallback(fallback)
        models = [model]
        if is_given(fallback_models):
            models.extend(item["model"] for item in fallback_models)
        aligned_transcript: Literal["word", False] = (
            "word" if all(_aligned_transcript_for_model(item) for item in models) else False
        )

        # chat_context follows the model's native support; the session decides whether to forward
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,
                interim_results=True,
                diarization=diarization_enabled,
                aligned_transcript=aligned_transcript,
                offline_recognize=False,
                keyterms=_keyterms_extra_for_model(model) is not None,
                chat_context=_supports_agent_context_carryover(model),
            ),
        )

        lk_base_url = base_url if is_given(base_url) else get_default_inference_url()

        lk_api_key = (
            api_key
            if is_given(api_key)
            else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
        )
        if not lk_api_key:
            raise ValueError(
                "api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
            )

        lk_api_secret = (
            api_secret
            if is_given(api_secret)
            else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
        )
        if not lk_api_secret:
            raise ValueError(
                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
            )

        self._opts = STTOptions(
            model=model,
            language=LanguageCode(language) if isinstance(language, str) else language,
            encoding=encoding if is_given(encoding) else DEFAULT_ENCODING,
            sample_rate=resolved_sample_rate,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            extra_kwargs=dict(extra_kwargs) if is_given(extra_kwargs) else {},
            fallback=fallback_models,
            conn_options=conn_options if is_given(conn_options) else DEFAULT_API_CONNECT_OPTIONS,
        )

        self._session = http_session
        self._vad = vad
        # a default sample rate follows the model; an explicit one is the caller's choice
        self._explicit_sample_rate = is_given(sample_rate)
        self._session_keyterms: list[str] = []  # framework-managed; merged into extra_kwargs
        self._streams = weakref.WeakSet[SpeechStream]()

    @classmethod
    def from_model_string(cls, model: str) -> STT:
        """Create a STT instance from a model string

        Args:
            model (str): STT model to use, in "provider/model[:language]" format

        Returns:
            STT: STT instance
        """
        model_name, language = _parse_model_string(model)
        return cls(model=model_name, language=language)

    @property
    def model(self) -> str:
        return self._opts.model if is_given(self._opts.model) else "unknown"

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

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()
        return self._session

    async def _recognize_impl(
        self,
        buffer: utils.AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        raise NotImplementedError(
            "LiveKit Inference STT does not support batch recognition, use stream() instead"
        )

    def stream(
        self,
        *,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        """Create a streaming transcription session."""
        options = self._sanitize_options(language=language)
        stream = SpeechStream(
            stt=self,
            opts=options,
            conn_options=conn_options,
            vad_instance=self._vad,
        )
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[STTLanguages | str] = NOT_GIVEN,
        extra: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> None:
        """Update STT configuration options."""
        # a ":language" suffix belongs to the model that carries it, so it is applied to
        # this STT but not forced onto a live stream that keeps an earlier model
        explicit_language = language
        if is_given(model):
            # Mirror __init__: strip ":language" suffix and apply if not overridden.
            if isinstance(model, str):
                parsed_model, parsed_language = _parse_model_string(model)
                model = parsed_model
                if is_given(parsed_language) and not is_given(language):
                    language = parsed_language

            self._opts.model = model
            self._vad = _resolve_vad_for_model(model, self._vad)
            if not self._explicit_sample_rate:
                # the default rate follows the model; a live stream's own rate stays
                # fixed at its creation, but streams created from now on must work
                self._opts.sample_rate = _required_sample_rate(model) or DEFAULT_SAMPLE_RATE
            _warn_if_sample_rate_refused(model, self._opts.sample_rate)
            models = [self._opts.model]
            if is_given(self._opts.fallback):
                models.extend(item["model"] for item in self._opts.fallback)
            self._capabilities = replace(
                self._capabilities,
                keyterms=_keyterms_extra_for_model(self._opts.model) is not None,
                chat_context=_supports_agent_context_carryover(self._opts.model),
                aligned_transcript=(
                    "word" if all(_aligned_transcript_for_model(item) for item in models) else False
                ),
            )
        if is_given(language):
            self._opts.language = LanguageCode(language)
        if is_given(extra):
            # keep this a pure user baseline: framework session keyterms are merged in
            # per stream at forward time, so a later clear can actually remove them
            self._opts.extra_kwargs.update(extra)
            self._capabilities = replace(
                self._capabilities,
                diarization=_diarization_enabled(self._opts.extra_kwargs),
            )

        for stream in self._streams:
            stream.update_options(model=model, language=explicit_language, extra=extra)

    def _update_session_keyterms(self, keyterms: list[str]) -> None:
        if keyterms == self._session_keyterms:
            return
        keyterm_extra = _keyterms_extra_for_model(
            self._opts.model, extra_kwargs=self._opts.extra_kwargs, session_keyterms=keyterms
        )
        if keyterm_extra is None:
            super()._update_session_keyterms(keyterms)  # warn-and-skip for unsupported models
            return

        self._session_keyterms = list(keyterms)
        # Each stream formats the terms for the model it is actually running (a refused
        # switch can leave it on a different provider than the parent). inference applies
        # extra live via session.update, so a speaking stream defers to END_OF_SPEECH
        # because the gateway may reconnect upstream when the keyterms change.
        for stream in self._streams:
            stream._on_session_keyterms_changed()

    def _push_conversation_item(self, ev: ConversationItemAddedEvent) -> None:
        if (
            (chat_item := ev.item).type == "message"
            and chat_item.role == "assistant"
            and (text := chat_item.text_content)
        ):
            if len(text) > _ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS:
                logger.debug(
                    "truncating agent_context carryover from %d to %d chars",
                    len(text),
                    _ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS,
                )
                text = text[-_ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS:]
            self.update_options(extra={"agent_context": text})

    def _sanitize_options(
        self, *, language: NotGivenOr[STTLanguages | str] = NOT_GIVEN
    ) -> STTOptions:
        """Create a sanitized copy of options with language override if provided."""
        options = replace(self._opts)
        options.extra_kwargs = dict(options.extra_kwargs)

        if is_given(language):
            options.language = LanguageCode(language)

        return options

class TTS(tts.TTS):
    @overload
    def __init__(
        self,
        model: CartesiaModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[CartesiaOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: DeepgramModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[DeepgramOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: RimeModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[RimeOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: InworldModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[InworldOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: XaiModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[XaiOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: FishAudioModels,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[FishAudioOptions] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: str,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        pass

    def __init__(
        self,
        model: TTSModels | str,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        encoding: NotGivenOr[TTSEncoding] = NOT_GIVEN,
        sample_rate: NotGivenOr[int] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        extra_kwargs: NotGivenOr[
            dict[str, Any]
            | CartesiaOptions
            | DeepgramOptions
            | RimeOptions
            | InworldOptions
            | XaiOptions
            | FishAudioOptions
        ] = NOT_GIVEN,
        fallback: NotGivenOr[list[FallbackModelType] | FallbackModelType] = NOT_GIVEN,
        conn_options: NotGivenOr[APIConnectOptions] = NOT_GIVEN,
    ) -> None:
        """Livekit Cloud Inference TTS

        Args:
            model (TTSModels | str): TTS model to use, in "provider/model[:voice]" format
            voice (str, optional): Voice to use, use a default one if not provided
            language (str, optional): Language of the TTS model.
            encoding (TTSEncoding, optional): Encoding of the TTS model.
            sample_rate (int, optional): Sample rate of the TTS model.
            base_url (str, optional): LIVEKIT_URL, if not provided, read from environment variable.
            api_key (str, optional): LIVEKIT_API_KEY, if not provided, read from environment variable.
            api_secret (str, optional): LIVEKIT_API_SECRET, if not provided, read from environment variable.
            http_session (aiohttp.ClientSession, optional): HTTP session to use.
            extra_kwargs (dict, optional): Extra kwargs to pass to the TTS model.
            fallback (FallbackModelType, optional): Fallback models - either a list of model names,
                a list of FallbackModel instances.
            conn_options (APIConnectOptions, optional): Connection options for request attempts.
        """
        sample_rate = sample_rate if is_given(sample_rate) else DEFAULT_SAMPLE_RATE

        # Parse voice from model string if provided: "provider/model:voice"
        if isinstance(model, str):
            parsed_model, parsed_voice = _parse_model_string(model)
            model = parsed_model
            if parsed_voice is not None and not is_given(voice):
                voice = parsed_voice

        resolved_extra_kwargs = dict(extra_kwargs) if is_given(extra_kwargs) else {}
        super().__init__(
            capabilities=tts.TTSCapabilities(
                streaming=True,
                aligned_transcript=_has_aligned_transcript(model, resolved_extra_kwargs),
            ),
            sample_rate=sample_rate,
            num_channels=1,
        )

        lk_base_url = base_url if is_given(base_url) else get_default_inference_url()

        lk_api_key = (
            api_key
            if is_given(api_key)
            else os.getenv("LIVEKIT_INFERENCE_API_KEY", os.getenv("LIVEKIT_API_KEY", ""))
        )
        if not lk_api_key:
            raise ValueError(
                "api_key is required, either as argument or set LIVEKIT_API_KEY environmental variable"
            )

        lk_api_secret = (
            api_secret
            if is_given(api_secret)
            else os.getenv("LIVEKIT_INFERENCE_API_SECRET", os.getenv("LIVEKIT_API_SECRET", ""))
        )
        if not lk_api_secret:
            raise ValueError(
                "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
            )

        fallback_models: NotGivenOr[list[FallbackModel]] = NOT_GIVEN
        if is_given(fallback):
            fallback_models = _normalize_fallback(fallback)

        self._opts = _TTSOptions(
            model=model,
            voice=voice,
            language=LanguageCode(language) if isinstance(language, str) else language,
            encoding=encoding if is_given(encoding) else DEFAULT_ENCODING,
            sample_rate=sample_rate,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            extra_kwargs=resolved_extra_kwargs,
            fallback=fallback_models,
            conn_options=conn_options if is_given(conn_options) else DEFAULT_API_CONNECT_OPTIONS,
        )
        self._session = http_session
        self._pool = utils.ConnectionPool[_TTSConnection](
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
            max_session_duration=300,
            mark_refreshed_on_get=True,
        )
        self._streams = weakref.WeakSet[SynthesizeStream]()

    class Markup(tts.TTS.Markup):
        def __init__(self, gateway_tts: TTS) -> None:
            super().__init__(gateway_tts)
            self._gateway_tts = gateway_tts

        def _provider_key(self) -> str:
            model = self._gateway_tts._opts.model
            provider = model.split("/")[0]
            if provider == "inworld" and "tts-2" in model:
                return "inworld"
            elif provider == "inworld":
                return ""  # older inworld models don't support markup
            elif provider == "fishaudio" and "drama" in model:
                return "fishaudio_drama"  # free-form direction instead of fixed emotions
            return provider

        # llm_instructions / normalize / convert are inherited from the base Markup,
        # keyed on _provider_key() above.

    @classmethod
    def from_model_string(cls, model: str) -> TTS:
        """Create a TTS instance from a model string

        Args:
            model (str): TTS model to use, in "provider/model[:voice_id]" format

        Returns:
            TTS: TTS instance
        """
        model, voice = _parse_model_string(model)
        return cls(model=model, voice=voice if voice else NOT_GIVEN)

    @property
    def model(self) -> str:
        return self._opts.model

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

    async def _connect_ws(self, timeout: float) -> _TTSConnection:
        session = self._ensure_session()
        base_url = self._opts.base_url
        if base_url.startswith(("http://", "https://")):
            base_url = base_url.replace("http", "ws", 1)

        headers = {
            **get_inference_headers(),
            "Authorization": f"Bearer {create_access_token(self._opts.api_key, self._opts.api_secret)}",
        }
        session_id = headers.get(HEADER_SESSION_ID)
        ws = None
        try:
            ws = await asyncio.wait_for(
                session.ws_connect(
                    f"{base_url}/tts?{urlencode({'model': self._opts.model})}", headers=headers
                ),
                timeout,
            )
        except aiohttp.ClientResponseError as e:
            raise create_api_error_from_http(e.message, status=e.status) from e
        except asyncio.TimeoutError as e:
            raise APITimeoutError("LiveKit Inference TTS connection timed out.") from e
        except aiohttp.ClientConnectorError as e:
            raise APIConnectionError("failed to connect to LiveKit Inference TTS") from e

        params: dict[str, Any] = {
            "type": "session.create",
            "sample_rate": str(self._opts.sample_rate),
            "encoding": self._opts.encoding,
            "extra": self._opts.extra_kwargs,
        }

        if self._opts.voice:
            params["voice"] = self._opts.voice
        if self._opts.model:
            params["model"] = self._opts.model
        if self._opts.language:
            params["language"] = self._opts.language
        if self._opts.fallback:
            models = [
                {
                    "model": m.get("model"),
                    "voice": m.get("voice"),
                    "extra": m.get("extra_kwargs", {}),
                }
                for m in self._opts.fallback
            ]
            params["fallback"] = {"models": models}

        if self._opts.conn_options:
            params["connection"] = {
                "timeout": self._opts.conn_options.timeout,
                "retries": self._opts.conn_options.max_retry,
            }

        try:
            payload = json.dumps(params)
            await ws.send_str(payload)
        except Exception as e:
            await ws.close()
            raise APIConnectionError(
                "failed to send session.create message to LiveKit Inference TTS"
            ) from e

        return _TTSConnection(ws=ws, session_id=session_id)

    async def _close_ws(self, connection: _TTSConnection) -> None:
        await connection.ws.close()

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()

        return self._session

    def prewarm(self) -> None:
        self._pool.prewarm()

    def update_options(
        self,
        *,
        voice: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> None:
        """
        Args:
            voice (str, optional): Voice.
            model (TTSModels | str, optional): TTS model to use.
            language (str, optional): Language code for the TTS model.
            extra_kwargs (dict, optional): Extra kwargs to pass to the TTS model.
        """
        if is_given(model):
            self._opts.model = model
        if is_given(voice):
            self._opts.voice = voice
        if is_given(language):
            self._opts.language = LanguageCode(language)
        if is_given(extra_kwargs):
            self._opts.extra_kwargs.update(extra_kwargs)

        self._capabilities.aligned_transcript = _has_aligned_transcript(
            self._opts.model, self._opts.extra_kwargs
        )

    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.ChunkedStream:
        return self._synthesize_with_stream(text, conn_options=conn_options)

    def stream(
        self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> SynthesizeStream:
        stream = SynthesizeStream(tts=self, conn_options=conn_options)
        self._streams.add(stream)
        return stream

    async def aclose(self) -> None:
        for stream in list(self._streams):
            await stream.aclose()

        self._streams.clear()
        await self._pool.aclose()

class TurnDetector(_BaseStreamingTurnDetector):
    def __init__(
        self,
        *,
        version: NotGivenOr[TurnDetectorVersions] = NOT_GIVEN,
        unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        sample_rate: int = DEFAULT_SAMPLE_RATE,
        local_fallback: bool = True,
        http_session: aiohttp.ClientSession | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            local_fallback: Whether a ``v1`` detector may degrade to the local
                ``v1-mini`` model when the gateway fails. False keeps it cloud-only, and
                turns commit on the endpointing delay instead. The worker still preloads
                the mini weights (~244MB RSS) under ``forkserver`` unless
                ``LIVEKIT_AGENTS_PRELOAD_LOCAL_INFERENCE=0`` is set.
        """
        auto = not is_given(version)
        resolved_version: TurnDetectorVersions = (
            version
            if is_given(version)
            else ("v1" if (utils.is_hosted() or utils.is_dev_mode()) else "v1-mini")
        )
        resolved_model: TurnDetectorModels = (
            "turn-detector-v1" if resolved_version == "v1" else "turn-detector-v1-mini"
        )

        cloud_opts: _CloudTransportOptions | None = None

        if resolved_version == "v1":
            lk_base_url = utils.resolve_env_var(
                base_url,
                "LIVEKIT_INFERENCE_URL",
                default=get_default_inference_url(),
            )
            lk_api_key = utils.resolve_env_var(
                api_key, "LIVEKIT_INFERENCE_API_KEY", "LIVEKIT_API_KEY", default=""
            )
            lk_api_secret = utils.resolve_env_var(
                api_secret,
                "LIVEKIT_INFERENCE_API_SECRET",
                "LIVEKIT_API_SECRET",
                default="",
            )
            missing: list[str] = []
            if not lk_base_url:
                missing.append("LIVEKIT_INFERENCE_URL")
            if not lk_api_key:
                missing.append("LIVEKIT_API_KEY")
            if not lk_api_secret:
                missing.append("LIVEKIT_API_SECRET")
            if missing:
                if auto:
                    logger.warning(
                        "LIVEKIT_INFERENCE_URL is set but %s missing; "
                        "falling back to the turn-detector-v1-mini model",
                        ", ".join(missing),
                    )
                    resolved_model = "turn-detector-v1-mini"
                else:
                    raise ValueError(
                        f"TurnDetector(version='v1') requires "
                        f"{', '.join(missing)} (env or constructor argument)."
                    )
            else:
                cloud_opts = _CloudTransportOptions(
                    base_url=lk_base_url,
                    api_key=lk_api_key,
                    api_secret=lk_api_secret,
                    conn_options=conn_options,
                )

        opts = TurnDetectorOptions(
            sample_rate=sample_rate,
            thresholds=ThresholdOptions(resolved_model, unlikely_threshold, backchannel_threshold),
        )
        super().__init__(opts=opts)

        self._model: TurnDetectorModels = resolved_model
        self._cloud_opts = cloud_opts
        self._local_fallback = local_fallback
        self._http_session = http_session

        if not local_fallback and resolved_model == "turn-detector-v1-mini":
            logger.warning(
                "local_fallback=False has no effect on the %s model, which runs locally by design",
                resolved_model,
            )

        self._warn_threshold_override()

    @property
    def model(self) -> TurnDetectorModels:
        return self._model

    def describe_options(self) -> dict[str, Any]:
        """What the session report shows for this detector (``telemetry.DescribesOptions``):
        the model and where it runs, plus the threshold overrides when the user set any.
        Server-calibrated defaults are not repeated here; credentials and endpoints never."""
        options: dict[str, Any] = {
            "model": self.model,
            "provider": self.provider,
            "sample_rate": self._opts.sample_rate,
            "local_fallback": self._local_fallback,
        }
        thresholds = self._opts.thresholds
        if is_given(thresholds.overrides):
            options["threshold_overrides"] = thresholds.overrides
        if is_given(thresholds.backchannel_overrides):
            options["backchannel_threshold_overrides"] = thresholds.backchannel_overrides
        return options

    def _warn_threshold_override(self) -> None:
        thresholds = self._opts.thresholds
        if is_given(overrides := thresholds.overrides):
            logger.warning(
                "a non-default turn detection threshold was provided "
                "(unlikely_threshold=%s); the server provides calibrated defaults and "
                "overriding them may be suboptimal",
                overrides,
            )
        if is_given(bc_overrides := thresholds.backchannel_overrides):
            logger.warning(
                "a non-default backchannel threshold was provided "
                "(backchannel_threshold=%s); the server provides calibrated defaults and "
                "overriding them may be suboptimal",
                bc_overrides,
            )

    def update_options(
        self,
        *,
        unlikely_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
        backchannel_threshold: NotGivenOr[float | dict[LanguageCode | str, float]] = NOT_GIVEN,
    ) -> None:
        if is_given(unlikely_threshold):
            self._opts.thresholds.update_overrides(unlikely_threshold)
        if is_given(backchannel_threshold):
            self._opts.thresholds.update_backchannel_overrides(backchannel_threshold)
        self._warn_threshold_override()

    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> _BaseStreamingTurnDetectorStream:
        cloud_opts = (
            replace(self._cloud_opts, conn_options=conn_options)
            if self._cloud_opts is not None
            else None
        )

        transport: _StreamingTurnDetectionTransport
        if self._model == "turn-detector-v1":
            assert cloud_opts is not None, "turn-detector-v1 requires cloud_opts"
            transport = _CloudTransport(
                detector=self,
                opts=self._opts,
                cloud_opts=cloud_opts,
                http_session=self._http_session,
            )
        else:
            transport = _LocalTransport(opts=self._opts)

        return _BaseStreamingTurnDetectorStream(
            detector=self,
            opts=self._opts,
            transport=transport,
            model=self._model,
            local_fallback=self._local_fallback,
        )

class VAD(vad.VAD):
    """Voice Activity Detection backed by ``livekit-local-inference``.

    The native model singleton is loaded once at module import (via the
    pybind11 ``.so`` constructor); each stream allocates its own per-instance
    LSTM/context state. Each stream uses a dedicated one-worker executor by default.
    When an executor is provided, the caller remains responsible for shutting it down.
    """

    def __init__(
        self,
        *,
        model: VADModels = "silero",
        min_speech_duration: float = 0.05,
        min_silence_duration: float = 0.25,
        prefix_padding_duration: float = 0.5,
        max_buffered_speech: float = 60.0,
        activation_threshold: float = 0.5,
        deactivation_threshold: NotGivenOr[float] = NOT_GIVEN,
        executor: Executor | None = None,
    ) -> None:
        super().__init__(capabilities=vad.VADCapabilities(update_interval=0.032))
        if model != "silero":
            raise ValueError(f"Unknown VAD model: {model!r}. Supported: 'silero'.")
        if is_given(deactivation_threshold) and deactivation_threshold <= 0:
            raise ValueError("deactivation_threshold must be greater than 0")
        self._model = model
        self._executor = executor
        self._opts = _VADOptions(
            min_speech_duration=min_speech_duration,
            min_silence_duration=min_silence_duration,
            prefix_padding_duration=prefix_padding_duration,
            max_buffered_speech=max_buffered_speech,
            activation_threshold=activation_threshold,
            deactivation_threshold=deactivation_threshold
            if is_given(deactivation_threshold)
            else max(activation_threshold - 0.15, 0.01),
        )
        self._streams: weakref.WeakSet[_VADStream] = weakref.WeakSet()

    @property
    def model(self) -> str:
        return self._model

    @property
    def provider(self) -> str:
        return "livekit-local-inference"

    def stream(self) -> vad.VADStream:
        # Each stream owns its own _VADOptions snapshot so that
        # _VADStream.update_options() can read the prior value of
        # max_buffered_speech before mutating it. Sharing the dataclass would
        # let VAD.update_options() mutate the stream's view first, and the
        # stream would never observe an increase.
        stream = _VADStream(self, replace(self._opts), executor=self._executor)
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        min_speech_duration: NotGivenOr[float] = NOT_GIVEN,
        min_silence_duration: NotGivenOr[float] = NOT_GIVEN,
        prefix_padding_duration: NotGivenOr[float] = NOT_GIVEN,
        max_buffered_speech: NotGivenOr[float] = NOT_GIVEN,
        activation_threshold: NotGivenOr[float] = NOT_GIVEN,
        deactivation_threshold: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        if is_given(min_speech_duration):
            self._opts.min_speech_duration = min_speech_duration
        if is_given(min_silence_duration):
            self._opts.min_silence_duration = min_silence_duration
        if is_given(prefix_padding_duration):
            self._opts.prefix_padding_duration = prefix_padding_duration
        if is_given(max_buffered_speech):
            self._opts.max_buffered_speech = max_buffered_speech
        if is_given(activation_threshold):
            self._opts.activation_threshold = activation_threshold
        if is_given(deactivation_threshold):
            self._opts.deactivation_threshold = deactivation_threshold

        for stream in self._streams:
            stream.update_options(
                min_speech_duration=min_speech_duration,
                min_silence_duration=min_silence_duration,
                prefix_padding_duration=prefix_padding_duration,
                max_buffered_speech=max_buffered_speech,
                activation_threshold=activation_threshold,
                deactivation_threshold=deactivation_threshold,
            )

    @property
    def min_silence_duration(self) -> float | None:
        return self._opts.min_silence_duration