Module livekit.agents.voice

class Agent:
    def __init__(
        self,
        *,
        instructions: str | Instructions,
        id: str | None = None,
        chat_ctx: NotGivenOr[llm.ChatContext | None] = NOT_GIVEN,
        tools: list[llm.Tool | llm.Toolset] | None = None,
        stt: NotGivenOr[stt.STT | STTModels | str | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        turn_handling: NotGivenOr[TurnHandlingOptions] = NOT_GIVEN,
        tool_handling: NotGivenOr[ToolHandlingOptions] = NOT_GIVEN,
        llm: NotGivenOr[
            llm.LLM | llm.RealtimeModel | llm.DuplexModel | LLMModels | str | None
        ] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | TTSModels | str | None] = NOT_GIVEN,
        expressive: NotGivenOr[bool | ExpressiveOptions] = NOT_GIVEN,
        min_consecutive_speech_delay: NotGivenOr[float] = NOT_GIVEN,
        use_tts_aligned_transcript: NotGivenOr[bool] = NOT_GIVEN,
        # deprecated
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        mcp_servers: NotGivenOr[list[mcp.MCPServer] | None] = NOT_GIVEN,
    ) -> None:
        tools = tools or []
        if type(self) is Agent:
            self._id = "default_agent"
        else:
            self._id = id or misc.camel_to_snake_case(type(self).__name__)

        turn_handling = (
            _migrate_turn_handling(
                min_endpointing_delay=min_endpointing_delay,
                max_endpointing_delay=max_endpointing_delay,
                turn_detection=turn_detection,
                allow_interruptions=allow_interruptions,
            )
            if not is_given(turn_handling)
            else turn_handling
        )

        self._instructions = instructions
        self._tools = [*tools, *find_function_tools(self)]
        self._chat_ctx = chat_ctx.copy(tools=self._tools) if chat_ctx else ChatContext.empty()
        self._turn_detection = turn_handling.get("turn_detection", NOT_GIVEN)

        if isinstance(stt, str):
            stt = inference.STT.from_model_string(stt)

        if isinstance(llm, str):
            llm = inference.LLM.from_model_string(llm)

        if isinstance(tts, str):
            tts = inference.TTS.from_model_string(tts)

        self._stt = stt
        # a duplex model is wrapped on the way in, so nothing downstream sees one
        self._llm: NotGivenOr[LLM | RealtimeModel | None] = (
            DuplexRealtimeAdapter(llm) if isinstance(llm, DuplexModel) else llm
        )
        self._tts = tts
        self._vad = vad
        self._expressive: NotGivenOr[bool | ExpressiveOptions] = expressive

        self._allow_interruptions: NotGivenOr[bool] = NOT_GIVEN
        self._interruption_detection: NotGivenOr[Literal["adaptive", "vad"]] = NOT_GIVEN
        if is_given(raw_interruption := turn_handling.get("interruption", NOT_GIVEN)):
            if "enabled" in raw_interruption:
                self._allow_interruptions = raw_interruption["enabled"]
            if "mode" in raw_interruption:
                self._interruption_detection = raw_interruption["mode"]
        endpointing = turn_handling.get("endpointing", {})
        self._min_consecutive_speech_delay = min_consecutive_speech_delay
        self._use_tts_aligned_transcript = use_tts_aligned_transcript
        self._min_endpointing_delay = endpointing.get("min_delay", NOT_GIVEN)
        self._max_endpointing_delay = endpointing.get("max_delay", NOT_GIVEN)
        self._turn_handling = turn_handling
        # stored unresolved so the resolution chain can tell "set on agent" from "fall
        # back to session"; async_options absent on a given tool_handling means NOT_GIVEN
        self._async_tool_options = (
            tool_handling.get("async_options", NOT_GIVEN) if is_given(tool_handling) else NOT_GIVEN
        )

        if isinstance(mcp_servers, list) and len(mcp_servers) == 0:
            mcp_servers = None  # treat empty list as None (but keep NOT_GIVEN)

        self._mcp_servers = mcp_servers
        if self._mcp_servers:
            logger.warning(
                "passing MCP servers to AgentSession or Agent is deprecated "
                "and will be removed in a future version. Use `MCPToolset` instead."
            )
        self._activity: AgentActivity | None = None

    @property
    def id(self) -> str:
        return self._id

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

    @property
    def instructions(self) -> str | Instructions:
        """
        Returns:
            str: The core instructions that guide the agent's behavior.
        """
        return self._instructions

    @property
    def tools(self) -> list[llm.Tool | llm.Toolset]:
        """
        Returns:
            list[llm.Tool | llm.ToolSet]:
                A list of function tools available to the agent.
        """
        return self._tools.copy()

    @property
    def chat_ctx(self) -> llm.ChatContext:
        """
        Provides a read-only view of the agent's current chat context.

        Returns:
            llm.ChatContext: A read-only version of the agent's conversation history.

        See Also:
            update_chat_ctx: Method to update the internal chat context.
        """
        return _ReadOnlyChatContext(self._chat_ctx.items)

    @property
    def interruption_detection(self) -> NotGivenOr[Literal["adaptive", "vad"]]:
        return self._interruption_detection

    @property
    def audio_recognition(self) -> AudioRecognition:
        """Access the audio recognition system for this agent.

        The only public member is ``stt_context`` — live speaker metadata from the
        STT stream.

        Raises:
            RuntimeError: If the agent is not running.
        """
        activity = self._get_activity_or_raise()
        assert activity._audio_recognition is not None
        return activity._audio_recognition

    async def update_instructions(self, instructions: str) -> None:
        """
        Updates the agent's instructions.

        If the agent is running in realtime mode, this method also updates
        the instructions for the ongoing realtime session.

        Args:
            instructions (str):
                The new instructions to set for the agent.

        Raises:
            llm.RealtimeError: If updating the realtime session instructions fails.
        """
        if self._activity is None:
            self._instructions = instructions
            return

        await self._activity.update_instructions(instructions)

    async def update_tools(self, tools: list[llm.Tool | llm.Toolset]) -> None:
        """
        Updates the agent's available function tools.

        If the agent is running in realtime mode, this method also updates
        the tools for the ongoing realtime session.

        Args:
            tools (list[llm.Tool | llm.ToolSet]):
                The new list of function tools available to the agent.

        Raises:
            llm.RealtimeError: If updating the realtime session tools fails.
        """
        valid_tools: list[llm.Tool | llm.Toolset] = []
        for tool in tools:
            if isinstance(tool, (llm.Tool, llm.Toolset)):
                valid_tools.append(tool)
            elif resolved_tool := llm.tool_context._resolve_wrapped_tool(tool):
                valid_tools.append(resolved_tool)
            else:
                raise TypeError(f"Invalid tool type: {type(tool)}. Expected Tool or ToolSet.")

        tools = valid_tools
        if self._activity is None:
            self._tools = list({tool.id: tool for tool in tools}.values())
            self._chat_ctx = self._chat_ctx.copy(tools=self._tools)
            return

        await self._activity.update_tools(tools)

    async def update_chat_ctx(
        self, chat_ctx: llm.ChatContext, *, exclude_invalid_function_calls: bool = True
    ) -> None:
        """
        Updates the agent's chat context.

        If the agent is running in realtime mode, this method also updates
        the chat context for the ongoing realtime session.

        Args:
            chat_ctx (llm.ChatContext):
                The new or updated chat context for the agent.
            exclude_invalid_function_calls (bool): Whether to exclude function calls
                and outputs not from the agent's tools.

        Raises:
            llm.RealtimeError: If updating the realtime session chat context fails.
        """
        if self._activity is None:
            self._chat_ctx = chat_ctx.copy(
                tools=self._tools if exclude_invalid_function_calls else NOT_GIVEN
            )
            return

        await self._activity.update_chat_ctx(
            chat_ctx, exclude_invalid_function_calls=exclude_invalid_function_calls
        )

    def update_options(
        self,
        *,
        stt: NotGivenOr[stt.STT | STTModels | str | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        llm: NotGivenOr[
            llm.LLM | llm.RealtimeModel | llm.DuplexModel | LLMModels | str | None
        ] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | TTSModels | str | None] = NOT_GIVEN,
        expressive: NotGivenOr[bool | ExpressiveOptions] = NOT_GIVEN,
    ) -> None:
        """Swap the STT, VAD, LLM, or TTS on this agent, or change its expressive setting.
        Only the options passed are changed.

        Useful for switching a component mid-call (e.g. a different STT language or TTS voice).
        Strings resolve to inference models like the constructor. Pass ``None`` to disable a
        model (overriding the session), matching ``Agent(stt=None)``. If the agent is running,
        the swap applies to the live pipeline. ``expressive`` overrides the session value and
        takes effect on the next reply; pass ``False`` to force it off for this agent.

        Raises:
            RuntimeError: When swapping to or from a ``RealtimeModel`` while the agent is
                running; use ``AgentSession.update_agent`` instead.
        """
        if isinstance(stt, str):
            stt = inference.STT.from_model_string(stt)
        if isinstance(llm, str):
            llm = inference.LLM.from_model_string(llm)
        if isinstance(tts, str):
            tts = inference.TTS.from_model_string(tts)

        if isinstance(llm, DuplexModel):
            llm = DuplexRealtimeAdapter(llm)

        if self._activity is None:
            # not running: replace stored config, applied on the next start
            if is_given(stt):
                self._stt = stt
            if is_given(vad):
                self._vad = vad
            if is_given(llm):
                self._llm = llm
            if is_given(tts):
                self._tts = tts
            if is_given(expressive):
                self._expressive = expressive
            return

        self._activity._update_models(new_stt=stt, new_vad=vad, new_llm=llm, new_tts=tts)
        if is_given(expressive):
            # after _update_models so a rejected model swap leaves expressive untouched;
            # resolved per turn (agent value over session), no live plumbing needed
            self._expressive = expressive

    # -- Pipeline nodes --
    # They can all be overriden by subclasses, by default they use the STT/LLM/TTS specified in the
    # constructor of the VoiceAgent

    async def on_enter(self) -> None:
        """Called when the task is entered"""
        pass

    async def on_exit(self) -> None:
        """Called when the task is exited"""
        pass

    async def on_user_turn_completed(
        self, turn_ctx: llm.ChatContext, new_message: llm.ChatMessage
    ) -> None:
        """Called when the user has finished speaking, and the LLM is about to respond

        This is a good opportunity to update the chat context or edit the new message before it is
        sent to the LLM.
        """
        pass

    async def on_user_turn_exceeded(self, ev: UserTurnExceededEvent) -> None:
        """Called when the user turn has exceeded the configured limit.

        The user has been speaking for too long without the agent successfully
        responding. By default, generates a reply using the current turn's
        transcript (previous turns are already in the chat context).

        Override to customize (e.g., use session.say() with a canned message,
        or skip the interruption entirely).
        """
        await self.session.generate_reply(
            user_input=ev.transcript,
            instructions=(
                "The user has been speaking too long without giving a chance to reply. "
                "Politely cut in with a short reply or notice. Keep it short since the user cannot interrupt it."
            ),
            allow_interruptions=False,
            tool_choice="none",
        )

    def stt_node(
        self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
    ) -> (
        AsyncIterable[stt.SpeechEvent | str]
        | Coroutine[Any, Any, AsyncIterable[stt.SpeechEvent | str]]
        | Coroutine[Any, Any, None]
    ):
        """
        A node in the processing pipeline that transcribes audio frames into speech events.

        By default, this node uses a Speech-To-Text (STT) capability from the current agent.
        If the STT implementation does not support streaming natively, a VAD (Voice Activity
        Detection) mechanism is required to wrap the STT.

        You can override this node with your own implementation for more flexibility (e.g.,
        custom pre-processing of audio, additional buffering, or alternative STT strategies).

        Args:
            audio (AsyncIterable[rtc.AudioFrame]): An asynchronous stream of audio frames.
            model_settings (ModelSettings): Configuration and parameters for model execution.

        Yields:
            stt.SpeechEvent: An event containing transcribed text or other STT-related data.
        """
        return Agent.default.stt_node(self, audio, model_settings)

    def llm_node(
        self,
        chat_ctx: llm.ChatContext,
        tools: list[llm.Tool],
        model_settings: ModelSettings,
    ) -> (
        AsyncIterable[llm.ChatChunk | str | FlushSentinel]
        | Coroutine[Any, Any, AsyncIterable[llm.ChatChunk | str | FlushSentinel]]
        | Coroutine[Any, Any, str]
        | Coroutine[Any, Any, llm.ChatChunk]
        | Coroutine[Any, Any, None]
    ):
        """
        A node in the processing pipeline that processes text generation with an LLM.

        By default, this node uses the agent's LLM to process the provided context. It may yield
        plain text (as `str`) for straightforward text generation, or `llm.ChatChunk` objects that
        can include text and optional tool calls. `ChatChunk` is helpful for capturing more complex
        outputs such as function calls, usage statistics, or other metadata.

        You can override this node to customize how the LLM is used or how tool invocations
        and responses are handled.

        Args:
            chat_ctx (llm.ChatContext): The context for the LLM (the conversation history).
            tools (list[FunctionTool]): A list of callable tools that the LLM may invoke.
            model_settings (ModelSettings): Configuration and parameters for model execution.

        Yields/Returns:
            str: Plain text output from the LLM.
            llm.ChatChunk: An object that can contain both text and optional tool calls.
        """
        return Agent.default.llm_node(self, chat_ctx, tools, model_settings)

    def transcription_node(
        self, text: AsyncIterable[str | TimedString], model_settings: ModelSettings
    ) -> (
        AsyncIterable[str | TimedString]
        | Coroutine[Any, Any, AsyncIterable[str | TimedString]]
        | Coroutine[Any, Any, None]
    ):
        """
        A node in the processing pipeline that finalizes transcriptions from text segments.

        This node can be used to adjust or post-process text coming from an LLM (or any other
        source) into a final transcribed form. For instance, you might clean up formatting, fix
        punctuation, or perform any other text transformations here.

        You can override this node to customize post-processing logic according to your needs.

        Args:
            text (AsyncIterable[str | TimedString]): An asynchronous stream of text segments.
            model_settings (ModelSettings): Configuration and parameters for model execution.

        Yields:
            str: Finalized or post-processed text segments.
        """
        return Agent.default.transcription_node(self, text, model_settings)

    def tts_node(
        self, text: AsyncIterable[str], model_settings: ModelSettings
    ) -> (
        AsyncIterable[rtc.AudioFrame]
        | Coroutine[Any, Any, AsyncIterable[rtc.AudioFrame]]
        | Coroutine[Any, Any, None]
    ):
        """
        A node in the processing pipeline that synthesizes audio from text segments.

        By default, this node converts incoming text into audio frames using the Text-To-Speech
        from the agent.
        If the TTS implementation does not support streaming natively, it uses a sentence tokenizer
        to split text for incremental synthesis.

        You can override this node to provide different text chunking behavior, a custom TTS engine,
        or any other specialized processing.

        Args:
            text (AsyncIterable[str]): An asynchronous stream of text segments to be synthesized.
            model_settings (ModelSettings): Configuration and parameters for model execution.

        Yields:
            rtc.AudioFrame: Audio frames synthesized from the provided text.
        """
        return Agent.default.tts_node(self, text, model_settings)

    def realtime_audio_output_node(
        self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
    ) -> (
        AsyncIterable[rtc.AudioFrame]
        | Coroutine[Any, Any, AsyncIterable[rtc.AudioFrame]]
        | Coroutine[Any, Any, None]
    ):
        """A node processing the audio from the realtime LLM session before it is played out."""
        return Agent.default.realtime_audio_output_node(self, audio, model_settings)

    def _get_activity_or_raise(self) -> AgentActivity:
        """Get the current activity context for this task (internal)"""
        if self._activity is None:
            raise RuntimeError("no activity context found, the agent is not running")

        return self._activity

    class default:
        @staticmethod
        async def stt_node(
            agent: Agent, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
        ) -> AsyncGenerator[stt.SpeechEvent, None]:
            """Default implementation for `Agent.stt_node`"""
            activity = agent._get_activity_or_raise()
            assert activity.stt is not None, "stt_node called but no STT node is available"

            wrapped_stt = activity.stt
            temporary_adapter: stt.StreamAdapter | None = None

            if not activity.stt.capabilities.streaming:
                if not activity.vad:
                    raise RuntimeError(
                        f"The STT ({activity.stt.label}) does not support streaming, add a VAD to the AgentTask/VoiceAgent to enable streaming"  # noqa: E501
                        "Or manually wrap your STT in a stt.StreamAdapter"
                    )

                temporary_adapter = stt.StreamAdapter(stt=wrapped_stt, vad=activity.vad)
                wrapped_stt = temporary_adapter

            try:
                conn_options = activity.session.conn_options.stt_conn_options
                async with wrapped_stt.stream(conn_options=conn_options) as stream:
                    _audio_input_started_at: float = (
                        activity._audio_recognition._input_started_at
                        if activity._audio_recognition is not None
                        and activity._audio_recognition._input_started_at is not None
                        else (
                            activity.session._recorder_io.recording_started_at
                            if activity.session._recorder_io
                            and activity.session._recorder_io.recording_started_at
                            else activity.session._started_at
                            if activity.session._started_at
                            else time.time()
                        )
                    )
                    stream.start_time_offset = time.time() - _audio_input_started_at

                    @utils.log_exceptions(logger=logger)
                    async def _forward_input() -> None:
                        async for frame in audio:
                            stream.push_frame(frame)

                    forward_task = asyncio.create_task(_forward_input())
                    try:
                        async for event in stream:
                            yield event
                    finally:
                        await utils.aio.cancel_and_wait(forward_task)
            finally:
                if temporary_adapter is not None:
                    await temporary_adapter.aclose()

        @staticmethod
        async def llm_node(
            agent: Agent,
            chat_ctx: llm.ChatContext,
            tools: list[llm.Tool],
            model_settings: ModelSettings,
        ) -> AsyncGenerator[llm.ChatChunk | str | FlushSentinel, None]:
            """Default implementation for `Agent.llm_node`"""
            activity = agent._get_activity_or_raise()
            assert activity.llm is not None, "llm_node called but no LLM node is available"
            assert isinstance(activity.llm, llm.LLM), (
                "llm_node should only be used with LLM (non-multimodal/realtime APIs) nodes"
            )

            tool_choice = model_settings.tool_choice if model_settings else NOT_GIVEN
            activity_llm = activity.llm

            conn_options = activity.session.conn_options.llm_conn_options
            async with activity_llm.chat(
                chat_ctx=chat_ctx, tools=tools, tool_choice=tool_choice, conn_options=conn_options
            ) as stream:
                async for chunk in stream:
                    yield chunk

        @staticmethod
        async def tts_node(
            agent: Agent,
            text: AsyncIterable[str],
            model_settings: ModelSettings,
        ) -> AsyncGenerator[rtc.AudioFrame, None]:
            """Default implementation for `Agent.tts_node`"""
            activity = agent._get_activity_or_raise()
            if activity.tts is None:
                raise RuntimeError(
                    "`tts_node` called but no TTS node is available. If audio output is not needed, disable it using "
                    "`session.output.set_audio_enabled(False)`."
                )

            expressive_active = activity._resolve_expressive_options() is not None
            wrapped_tts = activity.tts
            temporary_adapter: tts.StreamAdapter | None = None

            if not activity.tts.capabilities.streaming:
                temporary_adapter = tts.StreamAdapter(
                    tts=wrapped_tts,
                    sentence_tokenizer=tokenize.blingfire.SentenceTokenizer(
                        retain_format=True,
                        # markup only exists in the stream when expressive is active
                        xml_aware=expressive_active,
                    ),
                )
                wrapped_tts = temporary_adapter

            try:
                # Mark whether expressive is active for this synthesis, synchronously
                # just before stream() snapshots it. Doing it here (the single synthesis
                # choke point for both generate_reply and say()) scopes it to this turn
                # rather than leaving stale state on the instance. The provider's chunk
                # defaults then drive the TTS's input tokenizer.
                activity.tts._set_expressive(expressive_active)

                conn_options = activity.session.conn_options.tts_conn_options
                async with wrapped_tts.stream(conn_options=conn_options) as stream:

                    async def _forward_input() -> None:
                        async for chunk in text:
                            stream.push_text(chunk)

                        stream.end_input()

                    forward_task = asyncio.create_task(_forward_input())
                    try:
                        async for ev in stream:
                            yield ev.frame
                    finally:
                        await utils.aio.cancel_and_wait(forward_task)
            finally:
                if temporary_adapter is not None:
                    await temporary_adapter.aclose()

        @staticmethod
        async def transcription_node(
            agent: Agent, text: AsyncIterable[str | TimedString], model_settings: ModelSettings
        ) -> AsyncGenerator[str | TimedString, None]:
            """Default implementation for `Agent.transcription_node`"""
            async for delta in text:
                yield delta

        @staticmethod
        async def realtime_audio_output_node(
            agent: Agent, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
        ) -> AsyncGenerator[rtc.AudioFrame, None]:
            """Default implementation for `Agent.realtime_audio_output_node`"""
            activity = agent._get_activity_or_raise()
            assert activity.realtime_llm_session is not None, (
                "realtime_audio_output_node called but no realtime LLM session is available"
            )

            async for frame in audio:
                yield frame

    @property
    def realtime_llm_session(self) -> llm.RealtimeSession:
        """
        Retrieve the realtime LLM session associated with the current agent.

        Raises:
            RuntimeError: If the agent is not running or the realtime LLM session is not available
        """
        if (rt_session := self._get_activity_or_raise().realtime_llm_session) is None:
            raise RuntimeError("no realtime LLM session")

        return rt_session

    @property
    def duplex_session(self) -> llm.DuplexSession:
        """
        Retrieve the duplex session of the current agent, for provider-specific APIs.

        A duplex model is driven through an adapter that presents it as a realtime session; this
        returns the plugin's own session, where a provider puts what the abstraction does not carry.

        Raises:
            RuntimeError: If the agent is not running, or is not running on a duplex model
        """
        rt_session = self._get_activity_or_raise().realtime_llm_session
        if not isinstance(rt_session, _DuplexRealtimeSession):
            raise RuntimeError("no duplex session, this agent is not running a DuplexModel")

        return rt_session.duplex_session

    @property
    def turn_detection(self) -> NotGivenOr[TurnDetectionMode | None]:
        """
        Retrieves the turn detection mode for identifying conversational turns.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a turn detection,
        the session's turn detection mode will be used at runtime instead.

        Returns:
            NotGivenOr[TurnDetectionMode | None]: An optional turn detection mode for managing conversation flow.
        """  # noqa: E501
        return self._turn_detection

    @turn_detection.setter
    def turn_detection(self, value: TurnDetectionMode | None) -> None:
        self._turn_detection = value

        if self._activity is not None:
            self._activity.update_options(turn_detection=value)

    @property
    def stt(self) -> NotGivenOr[stt.STT | None]:
        """
        Retrieves the Speech-To-Text component for the agent.

        If this property was not set at Agent creation, but an ``AgentSession`` provides an STT component,
        the session's STT will be used at runtime instead.

        Returns:
            NotGivenOr[stt.STT | None]: An optional STT component.
        """  # noqa: E501
        return self._stt

    @property
    def llm(self) -> NotGivenOr[llm.LLM | llm.RealtimeModel | None]:
        """
        Retrieves the Language Model or RealtimeModel used for text generation.

        If this property was not set at Agent creation, but an ``AgentSession`` provides an LLM or RealtimeModel,
        the session's model will be used at runtime instead.

        Returns:
            NotGivenOr[llm.LLM | llm.RealtimeModel | None]: The language model for text generation.
        """  # noqa: E501
        return self._llm

    @property
    def tts(self) -> NotGivenOr[tts.TTS | None]:
        """
        Retrieves the Text-To-Speech component for the agent.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a TTS component,
        the session's TTS will be used at runtime instead.

        Returns:
            NotGivenOr[tts.TTS | None]: An optional TTS component for generating audio output.
        """  # noqa: E501
        return self._tts

    @property
    def expressive(self) -> NotGivenOr[bool | ExpressiveOptions]:
        """
        Retrieves the expressive TTS delivery setting for the agent.

        If this property was not set at Agent creation, the ``AgentSession``'s ``expressive``
        value will be used at runtime instead. When set, it overrides the session value for
        this agent's turns, matching how ``llm`` and ``tts`` overrides behave.

        Returns:
            NotGivenOr[bool | ExpressiveOptions]: Whether expressive delivery is enabled,
                or its configuration.
        """
        return self._expressive

    @property
    def mcp_servers(self) -> NotGivenOr[list[mcp.MCPServer] | None]:
        """
        Retrieves the list of Model Context Protocol (MCP) servers providing external tools.

        If this property was not set at Agent creation, but an ``AgentSession`` provides MCP servers,
        the session's MCP servers will be used at runtime instead.

        Returns:
            NotGivenOr[list[mcp.MCPServer]]: An optional list of MCP servers.
        """  # noqa: E501
        return self._mcp_servers

    @property
    def vad(self) -> NotGivenOr[vad.VAD | None]:
        """
        Retrieves the Voice Activity Detection component for the agent.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a VAD component,
        the session's VAD will be used at runtime instead.

        Returns:
            NotGivenOr[vad.VAD | None]: An optional VAD component for detecting voice activity.
        """  # noqa: E501
        return self._vad

    @property
    def allow_interruptions(self) -> NotGivenOr[bool]:
        """
        Indicates whether interruptions (e.g., stopping TTS playback) are allowed.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a value for
        allowing interruptions, the session's value will be used at runtime instead.

        Returns:
            NotGivenOr[bool]: Whether interruptions are permitted.
        """
        return self._allow_interruptions

    @property
    def min_endpointing_delay(self) -> NotGivenOr[float]:
        """
        Minimum time-in-seconds since the last detected speech before the agent
        declares the user’s turn complete. In VAD mode this effectively behaves
        like max(VAD silence, min_endpointing_delay); in STT mode it is applied
        after the STT end-of-speech signal, so it can be additive with the STT
        provider’s endpointing delay.

        If this property was set at Agent creation, it will be used at runtime instead of the session's value.
        """
        return self._min_endpointing_delay

    @property
    def max_endpointing_delay(self) -> NotGivenOr[float]:
        """
        Maximum time-in-seconds the agent will wait before terminating the turn.

        If this property was set at Agent creation, it will be used at runtime instead of the session's value.
        """
        return self._max_endpointing_delay

    @property
    def min_consecutive_speech_delay(self) -> NotGivenOr[float]:
        """
        Retrieves the minimum consecutive speech delay for the agent.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a value for
        the minimum consecutive speech delay, the session's value will be used at runtime instead.

        Returns:
            NotGivenOr[float]: The minimum consecutive speech delay.
        """
        return self._min_consecutive_speech_delay

    @property
    def use_tts_aligned_transcript(self) -> NotGivenOr[bool]:
        """
        Indicates whether to use TTS-aligned transcript as the input of
        the ``transcription_node``.

        If this property was not set at Agent creation, but an ``AgentSession`` provides a value for
        the use of TTS-aligned transcript, the session's value will be used at runtime instead.

        Returns:
            NotGivenOr[bool]: Whether to use TTS-aligned transcript.
        """
        return self._use_tts_aligned_transcript

    @property
    def session(self) -> AgentSession:
        """
        Retrieve the VoiceAgent associated with the current agent.

        Raises:
            RuntimeError: If the agent is not running
        """
        return self._get_activity_or_raise().session

class AgentFalseInterruptionEvent(BaseModel):
    type: Literal["agent_false_interruption"] = "agent_false_interruption"
    resumed: bool
    """Whether the false interruption was resumed automatically."""
    created_at: float = Field(default_factory=time.time)

    # deprecated
    message: ChatMessage | None = None
    extra_instructions: str | None = None

    def __getattribute__(self, name: str) -> Any:
        if name in ["message", "extra_instructions"]:
            logger.warning(
                f"AgentFalseInterruptionEvent.{name} is deprecated, automatic resume is now supported"
            )
        return super().__getattribute__(name)

class AgentSession(rtc.EventEmitter[EventTypes], Generic[Userdata_T]):
    @deprecate_params(
        {
            "min_endpointing_delay": "Use turn_handling=TurnHandlingOptions(...) instead",
            "max_endpointing_delay": "Use turn_handling=TurnHandlingOptions(...) instead",
            "false_interruption_timeout": "Use turn_handling=TurnHandlingOptions(...) instead",
            "resume_false_interruption": "Use turn_handling=TurnHandlingOptions(...) instead",
            "allow_interruptions": "Use turn_handling=TurnHandlingOptions(...) instead",
            "discard_audio_if_uninterruptible": "Use turn_handling=TurnHandlingOptions(...) instead",
            "min_interruption_duration": "Use turn_handling=TurnHandlingOptions(...) instead",
            "preemptive_generation": "Use turn_handling=TurnHandlingOptions(...) instead",
            "min_interruption_words": "Use turn_handling=TurnHandlingOptions(...) instead",
            "turn_detection": "Use turn_handling=TurnHandlingOptions(...) instead",
            "agent_false_interruption_timeout": "Use turn_handling=TurnHandlingOptions(...) instead",
            "keyterms_options": "Use stt_context_options=STTContextOptions(...) instead",
        },
        target_version="v2.0",
    )
    def __init__(
        self,
        *,
        stt: NotGivenOr[stt.STT | STTModels | str] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        llm: NotGivenOr[
            llm.LLM | llm.RealtimeModel | llm.DuplexModel | LLMModels | str
        ] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | TTSModels | str] = NOT_GIVEN,
        turn_handling: NotGivenOr[TurnHandlingOptions] = NOT_GIVEN,
        stt_context_options: NotGivenOr[STTContextOptions] = NOT_GIVEN,
        # Tool settings
        tools: NotGivenOr[list[llm.Tool | llm.Toolset]] = NOT_GIVEN,
        tool_handling: NotGivenOr[ToolHandlingOptions] = NOT_GIVEN,
        max_tool_steps: int = 3,
        # TTS settings
        use_tts_aligned_transcript: NotGivenOr[bool] = NOT_GIVEN,
        tts_text_transforms: NotGivenOr[Sequence[TextTransforms] | None] = NOT_GIVEN,
        min_consecutive_speech_delay: float = 0.0,
        expressive: bool | ExpressiveOptions = False,
        # Misc settings
        userdata: NotGivenOr[Userdata_T] = NOT_GIVEN,
        video_sampler: NotGivenOr[_VideoSampler | None] = NOT_GIVEN,
        aec_warmup_duration: NotGivenOr[float | None] = NOT_GIVEN,
        ivr_detection: bool = False,
        user_away_timeout: float | None = 15.0,
        transcription_timeout: float | None = None,
        session_close_transcript_timeout: float = 2.0,
        # Runtime settings
        conn_options: NotGivenOr[SessionConnectOptions] = NOT_GIVEN,
        loop: asyncio.AbstractEventLoop | None = None,
        # deprecated
        preemptive_generation: NotGivenOr[bool] = NOT_GIVEN,
        min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        false_interruption_timeout: NotGivenOr[float | None] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode] = NOT_GIVEN,
        discard_audio_if_uninterruptible: NotGivenOr[bool] = NOT_GIVEN,
        min_interruption_duration: NotGivenOr[float] = NOT_GIVEN,
        min_interruption_words: NotGivenOr[int] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        resume_false_interruption: NotGivenOr[bool] = NOT_GIVEN,
        agent_false_interruption_timeout: NotGivenOr[float | None] = NOT_GIVEN,
        mcp_servers: NotGivenOr[list[mcp.MCPServer]] = NOT_GIVEN,
        keyterms_options: NotGivenOr[KeytermsOptions] = NOT_GIVEN,
    ) -> None:
        """`AgentSession` is the LiveKit Agents runtime that glues together
        media streams, speech/LLM components, and tool orchestration into a
        single real-time voice agent.

        It links audio, video, and text I/O with STT, VAD, TTS, and the LLM;
        handles turn detection, endpointing, interruptions, and multi-step
        tool calls; and exposes everything through event callbacks so you can
        focus on writing function tools and simple hand-offs rather than
        low-level streaming logic.

        Args:
            stt (stt.STT | str, optional): Speech-to-text backend.
            vad (vad.VAD, optional): Voice-activity detector. Defaults to the
                bundled silero VAD (``inference.VAD(model="silero")``) when
                omitted. Pass ``vad=None`` to opt out, or pass an explicit
                instance to customise options.
            llm (llm.LLM | llm.RealtimeModel | str, optional): LLM or RealtimeModel
            tts (tts.TTS | str, optional): Text-to-speech engine.
            tools (list[llm.FunctionTool | llm.RawFunctionTool], optional): List of
                tools shared by every agent in the agent session.
            tool_handling (ToolHandlingOptions, optional): Tool handling configuration.
                ``tool_handling["async_options"]`` holds prompt templates for ``ctx.update()`` /
                duplicate-handling / coalesced replies. Unspecified keys keep their defaults;
                can be overridden per-``Agent`` or per-``AsyncToolset``.
            mcp_servers (list[mcp.MCPServer], optional): List of MCP servers
                providing external tools for the agent to use.
            userdata (Userdata_T, optional): Arbitrary per-session user data.
            turn_handling (TurnHandlingOptions, optional): Configuration for turn handling.
            stt_context_options (STTContextOptions, optional): Conversation-aware context for the
                STT: static ``keyterms`` plus ``keyterm_detection`` for STTs that accept a term
                list, and ``forward_chat_context`` (on by default) that forwards conversation turns
                to STTs that consume context directly. Applied where the STT supports it, ignored
                otherwise.
            keyterms_options (KeytermsOptions, optional): Deprecated, use ``stt_context_options``
                instead. Its ``keyterms``/``keyterm_detection`` keys map onto the new option.
            max_endpointing_delay (float): Maximum time-in-seconds the agent
                will wait before terminating the turn. Default ``3.0`` s.
            max_tool_steps (int): Maximum consecutive tool calls per LLM turn.
                Default ``3``.
            video_sampler (_VideoSampler, optional): Uses
                :class:`VoiceActivityVideoSampler` when *NOT_GIVEN*; that sampler
                captures video at ~1 fps while the user is speaking and ~0.3 fps
                when silent by default.
            min_consecutive_speech_delay (float, optional): The minimum delay between
                consecutive speech. Default ``0.0`` s.
            use_tts_aligned_transcript (bool, optional): Whether to use TTS-aligned
                transcript as the input of the ``transcription_node``. Only applies
                if ``TTS.capabilities.aligned_transcript`` is ``True`` or ``streaming``
                is ``False``. When NOT_GIVEN, it's disabled.
            tts_text_transforms (Sequence[TextTransforms], optional): The transforms to apply
                to the tts input text, available built-in transforms: ``"filter_markdown"``, ``"filter_emoji"``.
                Set to ``None`` to disable. When NOT_GIVEN, all filters will be applied.
            expressive (bool | ExpressiveOptions, optional): Let the LLM steer how the
                agent sounds. When enabled, the provider's markup guide is injected into
                the LLM prompt so it can emit inline delivery tags (emotion, pacing,
                non-verbal sounds), which are rendered by the TTS and stripped from the
                transcript. Pass an :class:`ExpressiveOptions` dict to steer or override
                the injected instructions. Requires ``livekit.agents.inference.TTS`` with
                a model that declares a markup dialect; it stays off otherwise.
                An ``expressive`` set on the active :class:`Agent` overrides this value.
                Default ``False``.
            ivr_detection (bool): Whether to detect if the agent is interacting with an IVR system.
                Default ``False``.
            conn_options (SessionConnectOptions, optional): Connection options for
                stt, llm, and tts.
            loop (asyncio.AbstractEventLoop, optional): Event loop to bind the
                session to. Falls back to :pyfunc:`asyncio.get_event_loop()`.
            user_away_timeout (float, optional): If set, set the user state as
                "away" after this amount of time after user and agent are silent.
                Defaults to ``15.0`` s, set to ``None`` to disable.
            transcription_timeout (float, optional): If set, emit a
                ``user_transcription_timeout`` event when VAD detects user speech
                during the user's turn but no non-empty final transcript arrives
                within this amount of time after the speech ends. This can happen
                because STT failed or because audio was intentionally withheld from
                STT, such as during AEC warmup or uninterruptible agent speech. Use
                it to prompt the user to repeat themselves. A non-empty final transcript
                satisfies the timeout for the current turn even if adaptive interruption
                detection later discards it as part of a backchannel. Requires both VAD
                and STT. Disabled by default.
            aec_warmup_duration (float, optional): The duration in seconds that the agent
                will ignore user's audio interruptions after the agent starts speaking.
                This is useful to prevent the agent from being interrupted by echo before AEC is ready.
                Defaults to ``3.0``, or ``None`` for outbound SIP calls.
            session_close_transcript_timeout (float, optional): Seconds to wait for the
                final STT transcript when closing the session (after audio is detached).
                Default ``2.0`` s (independent of ``commit_user_turn``'s ``transcript_timeout``).
            preemptive_generation (NotGivenOr[bool | PreemptiveGenerationOptions]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            min_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            max_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            false_interruption_timeout (NotGivenOr[float | None]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            turn_detection (NotGivenOr[TurnDetectionMode]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            discard_audio_if_uninterruptible (NotGivenOr[bool]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            min_interruption_duration (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            min_interruption_words (NotGivenOr[int]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            allow_interruptions (NotGivenOr[bool]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            resume_false_interruption (NotGivenOr[bool]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
            agent_false_interruption_timeout (NotGivenOr[float | None]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
        """
        super().__init__()
        self._loop = loop or asyncio.get_event_loop()
        self._video_sampler = (
            video_sampler
            if is_given(video_sampler)
            else VoiceActivityVideoSampler(speaking_fps=1.0, silent_fps=0.3)
        )

        turn_handling = (
            _migrate_turn_handling(
                min_endpointing_delay=min_endpointing_delay,
                max_endpointing_delay=max_endpointing_delay,
                false_interruption_timeout=false_interruption_timeout,
                turn_detection=turn_detection,
                discard_audio_if_uninterruptible=discard_audio_if_uninterruptible,
                min_interruption_duration=min_interruption_duration,
                min_interruption_words=min_interruption_words,
                allow_interruptions=allow_interruptions,
                resume_false_interruption=resume_false_interruption,
                agent_false_interruption_timeout=agent_false_interruption_timeout,
                preemptive_generation=preemptive_generation,
            )
            if not is_given(turn_handling)
            else turn_handling
        )

        raw_turn_detection: TurnDetectionMode | None = turn_handling.get(
            "turn_detection", inference.TurnDetector()
        )
        endpointing_overrides = turn_handling.get("endpointing") or EndpointingOptions()
        endpointing = _resolve_endpointing(endpointing_overrides, turn_detection=raw_turn_detection)
        interruption = _resolve_interruption(turn_handling.get("interruption"))
        preemptive_gen = _resolve_preemptive_generation(turn_handling.get("preemptive_generation"))

        # stt_context_options supersedes the deprecated keyterms_options; when both are set it wins
        if is_given(stt_context_options):
            stt_context = stt_context_options
        elif is_given(keyterms_options):
            stt_context = _stt_context_from_keyterms_options(keyterms_options)
        else:
            stt_context = None
        user_turn_limit = _resolve_user_turn_limit(turn_handling.get("user_turn_limit"))
        self._aec_warmup_duration_explicit = is_given(aec_warmup_duration)
        resolved_aec_warmup_duration = (
            aec_warmup_duration if is_given(aec_warmup_duration) else _DEFAULT_AEC_WARMUP_DURATION
        )

        # This is the "global" chat_context, it holds the entire conversation history
        self._chat_ctx = ChatContext.empty()
        self._opts = AgentSessionOptions(
            turn_handling=TurnHandlingOptions(
                endpointing=endpointing,
                interruption=interruption,
                turn_detection=raw_turn_detection,
                preemptive_generation=preemptive_gen,
                user_turn_limit=user_turn_limit,
            ),
            stt_context_options=_resolve_stt_context_options(stt_context),
            endpointing_overrides=endpointing_overrides,
            max_tool_steps=max_tool_steps,
            user_away_timeout=user_away_timeout,
            transcription_timeout=transcription_timeout,
            min_consecutive_speech_delay=min_consecutive_speech_delay,
            tts_text_transforms=(
                tts_text_transforms
                if is_given(tts_text_transforms)
                else DEFAULT_TTS_TEXT_TRANSFORMS
            ),
            ivr_detection=ivr_detection,
            use_tts_aligned_transcript=(
                use_tts_aligned_transcript if is_given(use_tts_aligned_transcript) else None
            ),
            aec_warmup_duration=resolved_aec_warmup_duration,
            session_close_transcript_timeout=session_close_transcript_timeout,
            recording_options=_RECORDING_ALL_OFF.copy(),
        )
        self._expressive: bool | ExpressiveOptions = expressive
        self._conn_options = conn_options or SessionConnectOptions()
        self._started = False

        if isinstance(stt, str):
            stt = inference.STT.from_model_string(stt)

        if isinstance(llm, str):
            llm = inference.LLM.from_model_string(llm)

        if isinstance(tts, str):
            tts = inference.TTS.from_model_string(tts)

        self._stt = stt or None
        self._using_default_vad = not is_given(vad)
        if not is_given(vad):
            vad = inference.VAD(model="silero")
        self._vad = vad or None
        # a duplex model is wrapped on the way in, so nothing downstream sees one
        self._llm: LLM | RealtimeModel | None = (
            DuplexRealtimeAdapter(llm) if isinstance(llm, DuplexModel) else (llm or None)
        )
        self._tts = tts or None

        # eagerly establish DNS/TLS to the LLM provider so the first inference
        # request doesn't pay connection setup costs
        if isinstance(self._llm, LLM):
            self._llm.prewarm(loop=self._loop)

        self._keyterm_detector = KeytermDetector(
            static_keyterms=self._opts.stt_context_options["keyterms"],
            options=self._opts.stt_context_options["keyterm_detection"],
        )

        self._turn_detection = raw_turn_detection
        # whether the user passed turn_detection, vs the eager TurnDetector() default
        self._turn_detection_explicit = "turn_detection" in turn_handling
        self._interruption_detection = interruption.get("mode", NOT_GIVEN)
        self._mcp_servers = mcp_servers or None
        if self._mcp_servers:
            logger.warning(
                "passing MCP servers to AgentSession or Agent is deprecated "
                "and will be removed in a future version. Use `MCPToolset` instead."
            )
        self._tools = tools if is_given(tools) else []
        self._async_tool_options = _resolve_async_tool_options(
            tool_handling.get("async_options") if is_given(tool_handling) else None
        )

        # unrecoverable error counts; stt resets on transcript, llm/tts on speaking
        self._stt_error_counts = 0
        self._llm_error_counts = 0
        self._tts_error_counts = 0

        # aec warmup: disable interruptions while AEC warms up
        self._aec_warmup_remaining = resolved_aec_warmup_duration or 0.0
        self._aec_warmup_timer: asyncio.TimerHandle | None = None

        # configurable IO
        self._input = io.AgentInput(
            self._on_video_input_changed,
            self._on_audio_input_changed,
            audio_enabled_cb=self._on_audio_enabled_changed,
        )
        self._output = io.AgentOutput(
            self._on_video_output_changed,
            self._on_audio_output_changed,
            self._on_text_output_changed,
        )

        self._forward_audio_atask: asyncio.Task[None] | None = None
        self._forward_video_atask: asyncio.Task[None] | None = None
        self._update_activity_atask: asyncio.Task[None] | None = None
        self._activity_lock = asyncio.Lock()
        self._lock = asyncio.Lock()

        # used to keep a reference to the room io
        self._room_io: room_io.RoomIO | None = None
        self._recorder_io: RecorderIO | None = None
        self._session_transport: SessionTransport | None = None
        self._session_transport_audio_input: TcpAudioInput | None = None
        self._session_transport_audio_output: TcpAudioOutput | None = None
        self._session_host: SessionHost | None = None

        self._agent: Agent | None = None
        self._activity: AgentActivity | None = None
        self._next_activity: AgentActivity | None = None
        self._user_state: UserState = "listening"
        self._agent_state: AgentState = "initializing"
        self._user_away_timer: asyncio.TimerHandle | None = None

        self._userdata: Userdata_T | None = userdata if is_given(userdata) else None
        self._closing_task: asyncio.Task[None] | None = None
        self._closing: bool = False
        self._job_context_cb_registered: bool = False

        # count of active `claim_user_turn` scopes. while > 0, `wait_for_idle`
        # is held open and `user_state` is pinned to "speaking"
        self._user_turn_claims: int = 0
        self._user_turn_released: asyncio.Event = asyncio.Event()
        self._user_turn_released.set()

        # count of active `_wait_for_idle_and_hold` scopes; while > 0, non-holder
        # `wait_for_idle` callers block until release. holder bypasses via contextvar.
        self._idle_holds: int = 0
        self._idle_released: asyncio.Event = asyncio.Event()
        self._idle_released.set()

        self._global_run_state: RunResult | None = None
        # guards keeping an active run open across `_wait_for_idle_and_hold` scopes
        self._foreground_guards: set[asyncio.Future[None]] = set()
        # TODO(theomonnom): need a better way to expose early assistant metrics
        self._early_assistant_metrics: MetricsReport | None = None
        # the latest user turn no agent speech has reported e2e_latency for yet
        self._unanswered_user_metrics: MetricsReport | None = None

        # trace
        self._user_speaking_span: trace.Span | None = None
        self._agent_speaking_span: trace.Span | None = None
        self._session_span: trace.Span | None = None
        self._root_span_context: otel_context.Context | None = None
        # event loop stalls seen while this session ran, summarised on the agent_session span
        self._loop_stall_count = 0
        self._loop_stall_total = 0.0
        self._loop_stall_max = 0.0
        # parent for the startup spans while start() runs; passed explicitly, never current
        self._session_start_context: otel_context.Context | None = None
        self._session_ctx_token: Token[otel_context.Context] | None = None

        self._recorded_events: list[AgentEvent] = []
        self._started_at: float | None = None
        self._usage_collector = ModelUsageCollector()
        self._redaction_enabled = False
        self._warned_realtime_audio_redaction = False

        # ivr and AMD
        self._ivr_activity: IVRActivity | None = None
        self._amd: AMD | None = None

    @property
    def amd(self) -> AMD | None:
        """The Answering Machine Detection (AMD) instance, or ``None`` if AMD is disabled."""
        return self._amd

    def on(self, event: EventTypes, callback: Callable | None = None) -> Callable:
        if event == "metrics_collected" and callback is not None:
            logger.warning(
                "metrics_collected is deprecated. "
                "Use session_usage_updated for usage tracking "
                "and ChatMessage.metrics for per-turn latency."
            )
        return super().on(event, callback)

    def emit(self, event: EventTypes, arg: AgentEvent) -> None:
        self._recorded_events.append(arg)
        super().emit(event, arg)

    @property
    def userdata(self) -> Userdata_T:
        if self._userdata is None:
            raise ValueError("AgentSession userdata is not set")

        return self._userdata

    @userdata.setter
    def userdata(self, value: Userdata_T) -> None:
        self._userdata = value

    @property
    def turn_detection(self) -> TurnDetectionMode | None:
        return self._turn_detection

    @property
    def mcp_servers(self) -> list[mcp.MCPServer] | None:
        return self._mcp_servers

    @property
    def input(self) -> io.AgentInput:
        return self._input

    @property
    def output(self) -> io.AgentOutput:
        return self._output

    @property
    def options(self) -> AgentSessionOptions:
        return self._opts

    @property
    def conn_options(self) -> SessionConnectOptions:
        return self._conn_options

    @property
    def history(self) -> llm.ChatContext:
        return self._chat_ctx

    @property
    def keyterms(self) -> list[str]:
        """The effective keyterms (user-defined + auto-detected) currently applied to the STT."""
        return self._keyterm_detector.keyterms

    @property
    def current_speech(self) -> SpeechHandle | None:
        return self._activity.current_speech if self._activity is not None else None

    @property
    def user_state(self) -> UserState:
        return self._user_state

    @property
    def agent_state(self) -> AgentState:
        return self._agent_state

    @property
    def current_agent(self) -> Agent:
        if self._agent is None:
            raise RuntimeError("VoiceAgent isn't running")

        return self._agent

    @property
    def tools(self) -> list[llm.Tool | llm.Toolset]:
        return self._tools

    @property
    def usage(self) -> AgentSessionUsage:
        """Returns usage summaries for this session, one per model/provider combination."""
        return AgentSessionUsage(model_usage=self._usage_collector.flatten())

    def run(
        self,
        *,
        user_input: str,
        input_modality: Literal["text", "audio"] = "text",
        output_type: type[Run_T] | None = None,
        output_options: NotGivenOr[RunOutputOptions | None] = NOT_GIVEN,
    ) -> RunResult[Run_T]:
        if self._global_run_state is not None and not self._global_run_state.done():
            raise RuntimeError("nested runs are not supported")

        run_state = RunResult(
            user_input=user_input,
            output_type=output_type,
            output_options=output_options,
            session=self,
        )
        self._global_run_state = run_state
        self.generate_reply(user_input=user_input, input_modality=input_modality)
        return run_state

    @overload
    async def start(
        self,
        agent: Agent,
        *,
        capture_run: Literal[True],
        room: NotGivenOr[rtc.Room] = NOT_GIVEN,
        room_options: NotGivenOr[room_io.RoomOptions] = NOT_GIVEN,
        session_host: NotGivenOr[bool] = NOT_GIVEN,
        record: bool | RecordingOptions = True,
        # deprecated
        room_input_options: NotGivenOr[room_io.RoomInputOptions] = NOT_GIVEN,
        room_output_options: NotGivenOr[room_io.RoomOutputOptions] = NOT_GIVEN,
    ) -> RunResult: ...

    @overload
    async def start(
        self,
        agent: Agent,
        *,
        capture_run: Literal[False] = False,
        room: NotGivenOr[rtc.Room] = NOT_GIVEN,
        room_options: NotGivenOr[room_io.RoomOptions] = NOT_GIVEN,
        session_host: NotGivenOr[bool] = NOT_GIVEN,
        record: bool | RecordingOptions = True,
        # deprecated
        room_input_options: NotGivenOr[room_io.RoomInputOptions] = NOT_GIVEN,
        room_output_options: NotGivenOr[room_io.RoomOutputOptions] = NOT_GIVEN,
    ) -> None: ...

    async def start(
        self,
        agent: Agent,
        *,
        capture_run: bool = False,
        room: NotGivenOr[rtc.Room] = NOT_GIVEN,
        room_options: NotGivenOr[room_io.RoomOptions] = NOT_GIVEN,
        session_host: NotGivenOr[bool] = NOT_GIVEN,
        record: NotGivenOr[bool | RecordingOptions] = NOT_GIVEN,
        # deprecated
        room_input_options: NotGivenOr[room_io.RoomInputOptions] = NOT_GIVEN,
        room_output_options: NotGivenOr[room_io.RoomOutputOptions] = NOT_GIVEN,
    ) -> RunResult | None:
        """Start the voice agent.

        Create a default RoomIO if the input or output audio is not already set.
        If the console flag is provided, start a ChatCLI.

        Args:
            capture_run: Whether to return a RunResult and capture the run result during session start.
            room: The room to use for input and output
            session_host: Whether a `RemoteSession` can drive and observe this session
                over the room. Defaults to True, or False when another session in the
                same job is already the primary. Set False when the room already has
                its agent and this session is only a participant in it, such as a
                client talking to that agent. Ignored under the console subcommand,
                where hosting is how the console reaches the session at all.
            room_input_options: Options for the room input
            room_output_options: Options for the room output
            record: Whether to record the audio, transcripts, traces, or logs
        """
        async with self._lock:
            if self._started:
                return None

            self._started_at = time.time()

            # configure observability first
            record_is_given = is_given(record)
            job_ctx = get_job_context(required=False)
            if not is_given(record):
                # defer to server-side setting for recording
                record = job_ctx.job.enable_recording if job_ctx else False

            self._opts.recording_options = _resolve_recording_options(record)  # type: ignore[arg-type]
            if self._text_only:
                self._opts.recording_options["audio"] = False

            is_primary = True
            if job_ctx:
                # set the primary session
                if job_ctx._primary_agent_session is None or job_ctx._primary_agent_session is self:
                    job_ctx._primary_agent_session = self
                else:
                    is_primary = False
                    if recording_enabled(self._opts.recording_options):
                        if record_is_given:
                            raise RuntimeError(
                                "Only one `AgentSession` can be the primary at a time. "
                                "If you want to ignore primary designation, "
                                "use session.start(record=False)."
                            )
                        else:
                            # auto-disable recording for non-primary sessions when record is not given
                            self._opts.recording_options = _resolve_recording_options(False)

                job_ctx.init_recording(self._opts.recording_options)

            self._redaction_enabled = bool(
                self._opts.recording_options["redaction"]
                or (job_ctx and job_ctx.job.enable_redaction)
            )

            # hosting needs the primary designation as before, and the caller's consent
            hosting = is_primary and (session_host if is_given(session_host) else True)

            # Under a text simulation the simulated user interacts over text
            # streams only: disable audio I/O here, and STT/TTS/VAD via
            # AgentActivity (both consult _text_only).
            if self._text_only:
                logger.info("text simulation: disabling STT/TTS/VAD and audio I/O")

            self._session_span = current_span = tracer.start_span("agent_session")
            # we detach here to avoid context issues since tokens need to be detached
            # in the same context as it was created
            if self._session_ctx_token is not None:
                otel_context.detach(self._session_ctx_token)
                self._session_ctx_token = None
            ctx = trace.set_span_in_context(current_span)
            self._session_ctx_token = otel_context.attach(ctx)

            self._recorded_events = []
            self._usage_collector = ModelUsageCollector()
            self._room_io = None
            self._recorder_io = None
            self._session_host = None

            self._closing = False
            self._root_span_context = otel_context.get_current()
            current_span = trace.get_current_span()
            current_span.set_attribute(trace_types.ATTR_AGENT_LABEL, agent.label)
            self._loop_stall_count = 0
            self._loop_stall_total = 0.0
            self._loop_stall_max = 0.0
            # the session is the convention's workflow: agent turns (`invoke_agent`),
            # inference (`chat`) and tool spans (`execute_tool`) nest underneath it
            gen_ai_telemetry.set_workflow_attributes(self._session_span, name="agent_session")

            # startup as one bar: room connect, participant wait, model prewarm, on_enter
            session_start_span = tracer.start_span("session_start")
            self._session_start_context = trace.set_span_in_context(session_start_span)

            self._agent = agent
            self._update_agent_state("initializing")

            tasks: list[asyncio.Task[None]] = []

            c = cli.AgentsConsole.get_instance()
            if c.enabled and not c.io_acquired:
                if self.input.audio is not None or self.output.audio is not None:
                    logger.warning(
                        "agent started with the console subcommand, but input.audio/output.audio "
                        "is already set, overriding..."
                    )

                c.acquire_io(loop=self._loop, session=self)

                if c._tcp_transport is not None:
                    self._session_host = SessionHost(
                        c._tcp_transport,
                        audio_input=c._tcp_audio_input,
                        audio_output=c._tcp_audio_output,
                    )
                    self._session_host.register_session(self)
            elif is_given(room) and not self._room_io:
                room_options = room_io.RoomOptions._ensure_options(
                    room_options,
                    room_input_options=room_input_options,
                    room_output_options=room_output_options,
                )
                room_options = copy.copy(room_options)  # shadow copy is enough

                if self._text_only:
                    room_options.audio_input = False
                    room_options.audio_output = False

                if self.input.audio is not None:
                    if room_options.audio_input:
                        logger.warning(
                            "RoomIO audio input is enabled but input.audio is already set, ignoring.."  # noqa: E501
                        )
                    room_options.audio_input = False

                if self.output.audio is not None:
                    if room_options.audio_output:
                        logger.warning(
                            "RoomIO audio output is enabled but output.audio is already set, ignoring.."  # noqa: E501
                        )
                    room_options.audio_output = False

                if self.output.transcription is not None:
                    if room_options.text_output:
                        logger.warning(
                            "RoomIO transcription output is enabled but output.transcription is already set, ignoring.."  # noqa: E501
                        )
                    room_options.text_output = False

                self._room_io = room_io.RoomIO(room=room, agent_session=self, options=room_options)
                # passed, not made current: RoomIO's tasks live for the whole session
                await self._room_io.start(trace_context=self._session_start_context)

                if hosting:
                    # only the primary session can have a session host
                    transport = RoomSessionTransport(room)
                    self._session_host = SessionHost(transport)
                    self._session_host.register_session(self)

                text_input_opts = room_options.get_text_input_options()
                if text_input_opts:
                    self._room_io.register_text_input(text_input_opts.text_input_cb)

            if job_ctx:
                # these aren't relevant during eval mode, as they require job context and/or room_io
                if self.input.audio and self.output.audio:
                    if self._opts.recording_options["audio"] or (c.enabled and c.record):
                        self._recorder_io = RecorderIO(agent_session=self)
                        self.input.audio = self._recorder_io.record_input(self.input.audio)
                        self.output.audio = self._recorder_io.record_output(self.output.audio)

                        if (c.enabled and c.record) or not c.enabled:
                            task = asyncio.create_task(
                                self._recorder_io.start(
                                    output_path=job_ctx.session_directory / "audio.ogg"
                                )
                            )
                            tasks.append(task)

                if self.options.ivr_detection:
                    tasks.append(
                        asyncio.create_task(self._start_ivr_detection(), name="_ivr_activity_start")
                    )

                current_span.set_attributes(
                    {
                        trace_types.ATTR_ROOM_NAME: job_ctx.room.name,
                        trace_types.ATTR_JOB_ID: job_ctx.job.id,
                        trace_types.ATTR_AGENT_NAME: job_ctx.job.agent_name,
                        # join keys shared with server, SIP and client traces
                        trace_types.ATTR_ROOM_SID: job_ctx.job.room.sid,
                        trace_types.ATTR_DISPATCH_ID: job_ctx.job.dispatch_id,
                        trace_types.ATTR_WORKER_ID: job_ctx._info.worker_id,
                        trace_types.ATTR_JOB_AGENT_ID: job_ctx.job.state.agent_id,
                    }
                )
                if self._room_io:
                    # automatically connect to the room when room io is used; room_connect
                    # finds session_start through the primary session (telemetry.session_context)
                    tasks.append(asyncio.create_task(job_ctx.connect(), name="_job_ctx_connect"))

                # session can be restarted, register the callbacks only once
                if not self._job_context_cb_registered:
                    job_ctx.add_shutdown_callback(
                        lambda: self._aclose_impl(reason=CloseReason.JOB_SHUTDOWN)
                    )
                    self._job_context_cb_registered = True

            run_state: RunResult | None = None
            if capture_run:
                if self._global_run_state is not None and not self._global_run_state.done():
                    raise RuntimeError("nested runs are not supported")

                run_state = RunResult(output_type=None)
                self._global_run_state = run_state

            # it is ok to await it directly, there is no previous task to drain.
            # _update_activity_task also watches on_enter on the run state: without it
            # the run completes as soon as the first speech does, dropping whatever
            # on_enter produces next — and never completes when on_enter says nothing.
            tasks.append(asyncio.create_task(self._update_activity_task(None, self._agent)))

            try:
                try:
                    await asyncio.gather(*tasks)
                finally:
                    await utils.aio.cancel_and_wait(*tasks)

                if self._session_host is not None:
                    await self._session_host.start()
            except Exception as e:
                trace_utils.record_exception(session_start_span, e)
                raise
            finally:
                session_start_span.end()
                self._session_start_context = None

            # important: no await should be done after this!

            if self.input.audio is not None:
                self._forward_audio_atask = asyncio.create_task(
                    self._forward_audio_task(), name="_forward_audio_task"
                )

            if self.input.video is not None:
                self._forward_video_atask = asyncio.create_task(
                    self._forward_video_task(), name="_forward_video_task"
                )

            self._started = True
            self._update_agent_state("listening")
            if self._room_io and self._room_io.subscribed_fut:

                def on_room_io_subscribed(_: asyncio.Future[None]) -> None:
                    if self._user_state == "listening" and self._agent_state == "listening":
                        self._set_user_away_timer()

                self._room_io.subscribed_fut.add_done_callback(on_room_io_subscribed)

            # log used IO
            def _collect_source(
                inp: io.AudioInput | io.VideoInput | None,
            ) -> list[io.AudioInput | io.VideoInput]:
                return [] if inp is None else [inp] + _collect_source(inp.source)

            def _collect_chain(
                out: io.TextOutput | io.VideoOutput | io.AudioOutput | None,
            ) -> list[io.VideoOutput | io.AudioOutput | io.TextOutput]:
                return [] if out is None else [out] + _collect_chain(out.next_in_chain)

            audio_input = _collect_source(self.input.audio)[::-1]
            video_input = _collect_source(self.input.video)[::-1]

            audio_output = _collect_chain(self.output.audio)
            video_output = _collect_chain(self.output.video)
            transcript_output = _collect_chain(self.output.transcription)

            logger.debug(
                "using audio io: %s -> `AgentSession` -> %s",
                " -> ".join([f"`{out.label}`" for out in audio_input]) or "(none)",
                " -> ".join([f"`{out.label}`" for out in audio_output]) or "(none)",
            )
            if (
                self._opts.interruption["resume_false_interruption"]
                and self.output.audio
                and not self.output.audio.can_pause
            ):
                logger.warning(
                    "resume_false_interruption is enabled but audio output does not support pause, it will be ignored",
                    extra={"audio_output": self.output.audio.label},
                )

            logger.debug(
                "using transcript io: `AgentSession` -> %s",
                " -> ".join([f"`{out.label}`" for out in transcript_output]) or "(none)",
            )

            if video_input or video_output:
                logger.debug(
                    "using video io: %s > `AgentSession` > %s",
                    " -> ".join([f"`{out.label}`" for out in video_input]) or "(none)",
                    " -> ".join([f"`{out.label}`" for out in video_output]) or "(none)",
                )

            if run_state:
                await run_state

            return run_state

    async def drain(self) -> None:
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        await self._activity.drain()

    @property
    def room_io(self) -> room_io.RoomIO:
        if not self._room_io:
            raise RuntimeError(
                "Cannot access room_io: the AgentSession was not started with a room."
            )

        return self._room_io

    def _close_soon(
        self,
        *,
        reason: CloseReason,
        drain: bool = False,
        error: (llm.LLMError | stt.STTError | tts.TTSError | llm.RealtimeModelError | None) = None,
    ) -> None:
        if self._closing_task:
            return
        self._closing_task = asyncio.create_task(
            self._aclose_impl(error=error, drain=drain, reason=reason)
        )

    def shutdown(self, *, drain: bool = True) -> None:
        self._close_soon(error=None, drain=drain, reason=CloseReason.USER_INITIATED)

    def _is_closing(self) -> bool:
        # _closing_task is set synchronously when close begins, earlier than _closing
        return self._closing_task is not None or self._closing

    @utils.log_exceptions(logger=logger)
    async def _aclose_impl(
        self,
        *,
        reason: CloseReason,
        drain: bool = False,
        error: (
            llm.LLMError
            | stt.STTError
            | tts.TTSError
            | llm.RealtimeModelError
            | inference.InterruptionDetectionError
            | None
        ) = None,
    ) -> None:
        # drain and on_exit run under the root span; the caller's context is restored after,
        # since aclose() may run in the caller's own task
        root_token: Token[otel_context.Context] | None = None
        if self._root_span_context:
            root_token = otel_context.attach(self._root_span_context)
        try:
            await self._aclose_locked(reason=reason, drain=drain, error=error)
        finally:
            if root_token is not None:
                otel_context.detach(root_token)

    async def _aclose_locked(
        self,
        *,
        reason: CloseReason,
        drain: bool,
        error: (
            llm.LLMError
            | stt.STTError
            | tts.TTSError
            | llm.RealtimeModelError
            | inference.InterruptionDetectionError
            | None
        ),
    ) -> None:
        async with self._lock:
            if not self._started:
                return

            # teardown as one bar under agent_session; drain and on_exit nest inside it
            close_span = tracer.start_span(
                "session_close",
                attributes={
                    trace_types.ATTR_CLOSE_REASON: reason.value,
                    trace_types.ATTR_CLOSE_DRAIN: drain,
                },
            )
            if error is not None:
                close_span.set_attribute(trace_types.ATTR_EXCEPTION_TYPE, error.type)
            close_token = otel_context.attach(trace.set_span_in_context(close_span))
            try:
                await self._teardown_activity(reason=reason, drain=drain)
            finally:
                close_span.end()
                # the rest of the teardown (close event, room io) is under agent_session
                otel_context.detach(close_token)

            if self._session_span:
                self._session_span.end()
                self._session_span = None

            self._started = False

            self.emit("close", CloseEvent(error=error, reason=reason))

            self._cancel_user_away_timer()
            self._user_state = "listening"
            self._agent_state = "initializing"
            self._stt_error_counts = 0
            self._llm_error_counts = 0
            self._tts_error_counts = 0
            self._root_span_context = None

            if self._global_run_state and not self._global_run_state.done():
                self._global_run_state._done_fut.set_exception(
                    RuntimeError(f"session closed: {error}" if error else "session closed")
                )

            if self._session_host:
                await self._session_host.aclose()
                self._session_host = None

            # close room io after close event is emitted
            if self._room_io:
                await self._room_io.aclose()
                self._room_io = None

        logger.debug("session closed", extra={"reason": reason.value, "error": error})

    async def _teardown_activity(self, *, reason: CloseReason, drain: bool) -> None:
        """The part of closing that runs under the ``session_close`` span."""
        self._closing = True
        self._cancel_user_away_timer()
        self._on_aec_warmup_expired()  # always clear aec warmup when closing the session

        if self._amd is not None:
            await self._amd.aclose()
            self._amd = None

        activity = self._activity
        while activity and isinstance(agent_task := activity.agent, AgentTask):
            # notify AgentTask to complete and wait it to resume the parent agent
            agent_task.cancel()
            await agent_task._wait_for_inactive()

            if old_agent := agent_task._old_agent:
                activity = old_agent._activity
            else:
                break

        if activity is not None:
            if not drain:
                try:
                    # force interrupt speeches when closing the session
                    await activity.interrupt(force=True)
                except RuntimeError:
                    # uninterruptible speech
                    pass
            await activity.drain()

            # wait any uninterruptible speech to finish
            if activity.current_speech:
                await activity.current_speech

            # detach the inputs and outputs
            self.input.audio = None
            self.input.video = None
            self.output.audio = None
            self.output.transcription = None

            if (
                reason != CloseReason.ERROR
                and (audio_recognition := activity._audio_recognition) is not None
            ):
                # wait for the user transcript to be committed
                audio_recognition._commit_user_turn(
                    audio_detached=True,
                    transcript_timeout=self._opts.session_close_transcript_timeout,
                )

            await activity.aclose()
        self._activity = None

        if self._agent_speaking_span:
            self._agent_speaking_span.end()
            self._agent_speaking_span = None

        if self._user_speaking_span:
            self._user_speaking_span.end()
            self._user_speaking_span = None

        if self._forward_audio_atask is not None:
            await utils.aio.cancel_and_wait(self._forward_audio_atask)

        if self._forward_video_atask is not None:
            await utils.aio.cancel_and_wait(self._forward_video_atask)

        if self._recorder_io:
            await self._recorder_io.aclose()

        if self._ivr_activity is not None:
            await self._ivr_activity.aclose()

        toolsets = [tool for tool in self._tools if isinstance(tool, llm.Toolset)]
        if toolsets:
            await asyncio.gather(
                *(toolset.aclose() for toolset in toolsets),
                return_exceptions=True,
            )

    async def aclose(self) -> None:
        await self._aclose_impl(reason=CloseReason.USER_INITIATED)

    def update_options(
        self,
        *,
        endpointing_opts: NotGivenOr[EndpointingOptions] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
        expressive: NotGivenOr[bool | ExpressiveOptions] = NOT_GIVEN,
        # deprecated
        min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        """
        Update the options for the agent session.

        Args:
            endpointing_opts (NotGivenOr[EndpointingOptions], optional): Endpointing options.
            turn_detection (NotGivenOr[TurnDetectionMode | None], optional): Strategy for deciding
                when the user has finished speaking. ``None`` reverts to automatic selection.
            keyterms (NotGivenOr[list[str]], optional): Replace the user-defined keyterms applied
                to the STT. Auto-detected keyterms are left untouched.
            expressive (NotGivenOr[bool | ExpressiveOptions], optional): Turn expressive TTS
                delivery on/off or change its options mid-session. Takes effect on the
                next reply. An ``expressive`` set on the active :class:`Agent` overrides the
                session value. When a turn runs with expressive off, markup left in past
                assistant messages is stripped from the chat history so the LLM doesn't
                imitate tags nothing downstream converts.
            min_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
            max_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
        """
        if is_given(expressive):
            # mypy can't narrow NotGiven out of the TypedDict union here
            self._expressive = cast("bool | ExpressiveOptions", expressive)
        if is_given(keyterms):
            self._keyterm_detector.set_static_keyterms(keyterms)
        if is_given(min_endpointing_delay) or is_given(max_endpointing_delay):
            logger.warning(
                "min_endpointing_delay and max_endpointing_delay are deprecated, "
                "use endpointing_opts instead"
            )
            endpointing_opts = EndpointingOptions()
            if is_given(min_endpointing_delay):
                endpointing_opts["min_delay"] = min_endpointing_delay
            if is_given(max_endpointing_delay):
                endpointing_opts["max_delay"] = max_endpointing_delay

        if is_given(endpointing_opts):
            if (mode := endpointing_opts.get("mode")) is not None:
                self._opts.endpointing["mode"] = mode
                self._opts.endpointing_overrides["mode"] = mode
            if (min_delay := endpointing_opts.get("min_delay")) is not None:
                self._opts.endpointing["min_delay"] = min_delay
                self._opts.endpointing_overrides["min_delay"] = min_delay
            if (max_delay := endpointing_opts.get("max_delay")) is not None:
                self._opts.endpointing["max_delay"] = max_delay
                self._opts.endpointing_overrides["max_delay"] = max_delay
            if (alpha := endpointing_opts.get("alpha")) is not None:
                self._opts.endpointing["alpha"] = alpha
                self._opts.endpointing_overrides["alpha"] = alpha

        if is_given(turn_detection):
            self._turn_detection = turn_detection
            self._turn_detection_explicit = turn_detection is not None

        if self._activity is not None:
            self._activity.update_options(
                endpointing_opts=(
                    self._opts.endpointing if is_given(endpointing_opts) else NOT_GIVEN
                ),
                turn_detection=turn_detection,
            )

    async def _start_ivr_detection(self, transcript: str | None = None) -> None:
        """Start IVR detection on this session.

        This method injects the DTMF tool and enables loop/silence detection,
        allowing the agent to navigate IVR phone trees. Safe to call after AMD resolves.

        Args:
            transcript (str | None, optional): The transcript to start IVR detection with.
        """
        if self._ivr_activity is not None:
            logger.warning("IVR detection already started, skipping")
            return

        self._ivr_activity = IVRActivity(self)
        self._tools.extend(self._ivr_activity.tools)
        await self._ivr_activity.start()
        if transcript is not None:
            logger.debug(
                "IVR detection started with transcript",
                extra={"lk.pii.transcript": transcript},
            )
            self._ivr_activity._on_user_input_transcribed(
                UserInputTranscribedEvent(transcript=transcript, is_final=True)
            )

    def say(
        self,
        text: str | AsyncIterable[str],
        *,
        audio: NotGivenOr[AsyncIterable[rtc.AudioFrame]] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        add_to_chat_ctx: bool = True,
    ) -> SpeechHandle:
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        run_state = self._global_run_state
        activity = self._next_activity if self._activity.scheduling_paused else self._activity

        if activity is None:
            raise RuntimeError("AgentSession is closing, cannot use say()")

        # attach to the session span if called outside of the AgentSession
        use_span: AbstractContextManager[trace.Span | None] = nullcontext()
        if trace.get_current_span() is trace.INVALID_SPAN and self._session_span is not None:
            use_span = tracer.use_span(self._session_span, end_on_exit=False)

        with use_span:
            handle = activity.say(
                text,
                audio=audio,
                allow_interruptions=allow_interruptions,
                add_to_chat_ctx=add_to_chat_ctx,
            )
            if run_state:
                run_state._watch_handle(handle)

        return handle

    def generate_reply(
        self,
        *,
        user_input: NotGivenOr[str | llm.ChatMessage] = NOT_GIVEN,
        instructions: NotGivenOr[str | Instructions] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
        tools: NotGivenOr[list[str]] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        chat_ctx: NotGivenOr[ChatContext] = NOT_GIVEN,
        input_modality: Literal["text", "audio"] = "text",
    ) -> SpeechHandle:
        """Generate a reply for the agent to speak to the user.

        Args:
            user_input (NotGivenOr[str | llm.ChatMessage], optional): The user's input that may influence the reply,
                such as answering a question.
            instructions (NotGivenOr[str], optional): Additional instructions for generating the reply.
            tool_choice (NotGivenOr[llm.ToolChoice], optional): Specifies the external tool to use when
                generating the reply. If generate_reply is invoked within a function_tool, defaults to "none".
            tools (NotGivenOr[list[str]], optional): List of tool IDs to make available for this response.
                When set, only the specified tools can be used. Tool IDs must match registered tools on the
                agent. For function tools, the ID is the function name (accessible via ``my_tool.id``).
                For toolsets, the ID is the one provided at construction (accessible via ``my_toolset.id``).
            allow_interruptions (NotGivenOr[bool], optional): Indicates whether the user can interrupt this speech.
            chat_ctx (NotGivenOr[ChatContext], optional): The chat context to use for generating the reply.
                Defaults to the chat context of the current agent if not provided.
            input_modality (Literal["text", "audio"], optional): The input mode to use for generating the reply.

        Returns:
            SpeechHandle: A handle to the generated reply.

        Note:
            ``await handle`` waits for the reply to finish and never raises; check
            ``handle.exception()`` for the failure instead.
        """  # noqa: E501
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        user_message = (
            llm.ChatMessage(role="user", content=[user_input])
            if isinstance(user_input, str)
            else user_input
        )

        run_state = self._global_run_state
        activity = self._next_activity if self._activity.scheduling_paused else self._activity

        if activity is None:
            raise RuntimeError("AgentSession is closing, cannot use generate_reply()")

        # attach to the session span if called outside of the AgentSession
        use_span: AbstractContextManager[trace.Span | None] = nullcontext()
        if trace.get_current_span() is trace.INVALID_SPAN and self._session_span is not None:
            use_span = tracer.use_span(self._session_span, end_on_exit=False)

        with use_span:
            handle = activity._generate_reply(
                user_message=user_message if user_message else None,
                instructions=instructions,
                tool_choice=tool_choice,
                tools=tools,
                allow_interruptions=allow_interruptions,
                chat_ctx=chat_ctx,
                input_details=InputDetails(modality=input_modality),
            )
            if run_state:
                run_state._watch_handle(handle)

        return handle

    def interrupt(self, *, force: bool = False) -> asyncio.Future[None]:
        """Interrupt the current speech generation.

        A queued speech created with ``allow_interruptions=False`` keeps playing,
        along with the ones behind it, unless ``force`` is set.

        Returns:
            An asyncio.Future that completes when the interruption is fully processed
            and chat context has been updated.

        Raises:
            RuntimeError: If the session isn't running, or if the speech currently
                playing disallows interruptions and ``force`` is False.
        """
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        return self._activity.interrupt(force=force)

    @asynccontextmanager
    async def _claim_user_turn(self) -> AsyncIterator[None]:
        """Declare a programmatic user-driven turn.

        Pins ``user_state`` to ``"speaking"`` and holds ``wait_for_idle``
        open until release. On release, ``user_state`` is re-derived from the
        audio path. Reentrant and session-scoped (survives handoff).

        Use in custom ``text_input_cb`` or any flow that drives a user turn
        across awaits.
        """
        first = self._user_turn_claims == 0
        self._user_turn_claims += 1
        if first:
            self._user_turn_released.clear()
            self._update_user_state("speaking", last_speaking_time=time.time())
        try:
            yield
        finally:
            self._user_turn_claims -= 1
            if self._user_turn_claims == 0:
                self._user_turn_released.set()
                activity = self._activity
                speaking = activity is not None and not activity._user_silence_event.is_set()
                self._update_user_state("speaking" if speaking else "listening")

    def clear_user_turn(self) -> None:
        # clear the transcription or input audio buffer of the user turn
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        self._activity.clear_user_turn()

    def commit_user_turn(
        self,
        *,
        transcript_timeout: float = 2.0,
        stt_flush_duration: float = 2.0,
        skip_reply: bool = False,
    ) -> asyncio.Future[str]:
        """Commit the user turn and generate a reply.

        Returns a future that resolves with the user's audio transcript once STT
        is complete and end-of-turn detection has been triggered.

        Args:
            transcript_timeout (float, optional): The timeout for the final transcript
                to be received after committing the user turn.
                Default ``2.0`` s. Increase this value if the STT is slow to respond.
            stt_flush_duration (float, optional): The duration of the silence to be appended to the STT
                to flush the buffer and generate the final transcript.
                Default ``2.0`` s.
            skip_reply (bool, optional): Whether to skip the reply generation after committing the user turn.

        Returns:
            asyncio.Future[str]: A future that resolves with the audio transcript.

        Raises:
            RuntimeError: If the AgentSession isn't running.
        """
        if self._activity is None:
            raise RuntimeError("AgentSession isn't running")

        return self._activity.commit_user_turn(
            transcript_timeout=transcript_timeout,
            stt_flush_duration=stt_flush_duration,
            skip_reply=skip_reply,
        )

    def update_agent(self, agent: Agent) -> None:
        self._agent = agent

        if self._started:
            # immediately block the old activity from accepting new user turns
            # during the transition window (before drain() formally pauses scheduling)
            if self._activity is not None:
                self._activity._new_turns_blocked = True

            self._update_activity_atask = task = asyncio.create_task(
                self._update_activity_task(self._update_activity_atask, self._agent),
                name="_update_activity_task",
            )
            run_state = self._global_run_state
            if run_state:
                # don't mark the RunResult as done, if there is currently an agent transition happening.  # noqa: E501
                # (used to make sure we're correctly adding the AgentHandoffResult before completion)  # noqa: E501
                run_state._watch_handle(task)

    async def wait_for_idle(self) -> AgentActivity:
        """Wait until the current activity is idle and return it. Re-targets on handoff.

        Raises ``ActivityClosedError`` if the session is closing,
        or ``RuntimeError`` if no activity has been started.
        """
        from .agent_activity import ActivityClosedError

        while True:
            if self._closing_task is not None:
                raise ActivityClosedError("session is closing")

            activity = self._activity
            if activity is None:
                raise RuntimeError("AgentSession has no active AgentActivity")

            try:
                await activity.wait_for_idle()
                return activity
            except ActivityClosedError:
                # handoff in flight — re-target to whatever's current now
                if self._activity is activity:
                    raise
                continue

    @asynccontextmanager
    async def _wait_for_idle_and_hold(
        self, *, run_state: RunResult | None = None
    ) -> AsyncIterator[AgentActivity]:
        """Wait for idle, then block other ``wait_for_idle`` callers until exit.

        ``run_state`` is the run that owns the turn of the caller. That run is guarded for
        the whole scope, since reaching idle is itself what completes the run: without the
        guard, whatever the scope does next lands after ``run()`` returned and goes
        unrecorded. A caller whose turn already ended guards nothing, so background work
        cannot hold a later run open.
        """
        from .agent_activity import _IdleHoldContextVar

        # registered before the wait below, which is what lets the run complete
        guard: asyncio.Future[None] | None = None
        if run_state is not None and run_state is self._global_run_state and not run_state.done():
            guard = asyncio.get_running_loop().create_future()
            run_state._watch_handle(guard)
            self._foreground_guards.add(guard)

        try:
            activity = await self.wait_for_idle()
            self._idle_holds += 1
            self._idle_released.clear()
            token = _IdleHoldContextVar.set(True)
            try:
                yield activity
            finally:
                _IdleHoldContextVar.reset(token)
                self._idle_holds -= 1
                if self._idle_holds == 0:
                    self._idle_released.set()
        finally:
            if guard is not None:
                self._foreground_guards.discard(guard)
                # the run re-checks from the done callback; releasing here rather than on
                # the way out of the inner block covers a floor that never arrives
                guard.set_result(None)

    async def _update_activity(
        self,
        agent: Agent,
        *,
        previous_activity: Literal["close", "pause"] = "close",
        new_activity: Literal["start", "resume"] = "start",
        blocked_tasks: list[asyncio.Task] | None = None,
        wait_on_enter: bool = True,
    ) -> None:
        async with self._activity_lock:
            if self._closing and new_activity == "start":
                # checked again after the drain below: closing may start while it's in flight
                logger.warning(
                    f"session is closing, skipping start activity of agent {agent.id}",
                )
                return

            # _update_activity is called directly sometimes, update for redundancy
            self._agent = agent

            if new_activity == "start":
                previous_agent = self._activity.agent if self._activity else None
                if agent._activity is not None and (
                    # allow updating the same agent that is running
                    agent is not previous_agent or previous_activity != "close"
                ):
                    raise RuntimeError("cannot start agent: an activity is already running")

                self._next_activity = AgentActivity(agent, self)
            elif new_activity == "resume":
                if agent._activity is None:
                    raise RuntimeError("cannot resume agent: no existing active activity to resume")

                self._next_activity = agent._activity

            if self._root_span_context is not None:
                # restore the root span context so on_exit, on_enter, and future turns
                # are direct children of the root span, not nested under a tool call.
                otel_context.attach(self._root_span_context)

            # one span for the handoff: the old agent's drain/pause and on_exit, then the new
            # one's start/resume and on_enter nest under it. It is made current only around
            # calls that spawn no long-lived tasks, and passed explicitly to the ones that do
            handoff_span: trace.Span | None = None
            handoff_ctx: otel_context.Context | None = None
            if self._activity is not None and self._next_activity is not None:
                handoff_span = tracer.start_span(
                    "update_agent",
                    attributes={
                        trace_types.ATTR_PREVIOUS_AGENT_LABEL: self._activity.agent.label,
                        trace_types.ATTR_AGENT_LABEL: self._next_activity.agent.label,
                    },
                )
                handoff_ctx = trace.set_span_in_context(handoff_span)
            under_handoff = (
                tracer.use_span(handoff_span, end_on_exit=False)
                if handoff_span is not None
                else contextlib.nullcontext()
            )

            reuse_resources: _ReusableResources | None = None
            try:
                previous_activity_v = self._activity
                if (activity := self._activity) is not None:
                    if previous_activity == "close":
                        activity._new_turns_blocked = True
                        with under_handoff:
                            reuse_resources = await activity.drain(new_activity=self._next_activity)
                            await activity.aclose()
                    elif previous_activity == "pause":
                        with under_handoff:
                            reuse_resources = await activity.pause(
                                blocked_tasks=blocked_tasks or [],
                                new_activity=self._next_activity,
                            )

                if self._closing and new_activity == "start":
                    # disallow starting a new activity when the session is closing
                    logger.warning(
                        f"session is closing, skipping {new_activity} activity of {self._next_activity.agent.id}",
                    )
                    if reuse_resources is not None:
                        await reuse_resources.cleanup()
                        reuse_resources = None
                    self._next_activity = None
                    self._activity = None
                    return

                self._activity = self._next_activity
                self._next_activity = None

                run_state = self._global_run_state
                handoff_item = AgentHandoff(
                    old_agent_id=(previous_activity_v.agent.id if previous_activity_v else None),
                    new_agent_id=self._activity.agent.id,
                )
                if run_state:
                    run_state._agent_handoff(
                        item=handoff_item,
                        old_agent=(previous_activity_v.agent if previous_activity_v else None),
                        new_agent=self._activity.agent,
                    )
                self._chat_ctx.insert(handoff_item)
                self.emit(
                    "conversation_item_added",
                    ConversationItemAddedEvent(item=handoff_item),
                )

                if new_activity == "start":
                    await self._activity.start(
                        reuse_resources=reuse_resources,
                        trace_context=(
                            handoff_ctx if handoff_ctx is not None else self._session_start_context
                        ),
                    )
                elif new_activity == "resume":
                    await self._activity.resume(
                        reuse_resources=reuse_resources, trace_context=handoff_ctx
                    )
            except BaseException as e:
                if handoff_span is not None and isinstance(e, Exception):
                    trace_utils.record_exception(handoff_span, e)
                if reuse_resources is not None:
                    await reuse_resources.cleanup()
                raise
            finally:
                if handoff_span is not None:
                    handoff_span.end()

        # move it outside the lock to allow calling _update_activity in on_enter of a new agent
        if wait_on_enter:
            assert self._activity._on_enter_task is not None
            await asyncio.shield(self._activity._on_enter_task)

    @utils.log_exceptions(logger=logger)
    async def _update_activity_task(
        self, old_task: asyncio.Task[None] | None, agent: Agent
    ) -> None:
        if old_task is not None:
            await old_task

        await self._update_activity(agent, wait_on_enter=False)

        # watch on_enter so the run captures its output without awaiting it
        if (activity := self._activity) is not None and activity._on_enter_task is not None:
            if (run_state := self._global_run_state) is not None and not run_state.done():
                run_state._watch_handle(activity._on_enter_task)

    def _emit_debug_message(self, payload: dict[str, Any]) -> None:
        """:meta private: internal — emit a debug/trace payload to the debugger/recorder."""
        st = Struct()
        ParseDict(payload, st)
        # super().emit bypasses AgentSession.emit's narrowed AgentEvent type;
        # debug messages ride the proto, not the Pydantic event union.
        super().emit("debug_message", agent_pb.DebugMessage(payload=st))

    def _on_error(
        self, error: llm.LLMError | stt.STTError | tts.TTSError | llm.RealtimeModelError
    ) -> None:
        if self._closing_task or error.recoverable:
            return

        if error.type == "stt_error":
            self._stt_error_counts += 1
            if self._stt_error_counts <= self.conn_options.max_unrecoverable_errors:
                return
        elif error.type == "llm_error":
            self._llm_error_counts += 1
            if self._llm_error_counts <= self.conn_options.max_unrecoverable_errors:
                return
        elif error.type == "tts_error":
            self._tts_error_counts += 1
            if self._tts_error_counts <= self.conn_options.max_unrecoverable_errors:
                return

        if isinstance(error.error, APIError):
            logger.error(f"AgentSession is closing due to unrecoverable error: {error.error}")
        else:
            logger.error(
                "AgentSession is closing due to unrecoverable error",
                exc_info=error.error,
            )

        def on_close_done(_: asyncio.Task[None]) -> None:
            self._closing_task = None

        self._closing_task = asyncio.create_task(
            self._aclose_impl(error=error, reason=CloseReason.ERROR)
        )
        self._closing_task.add_done_callback(on_close_done)

    @utils.log_exceptions(logger=logger)
    async def _forward_audio_task(self) -> None:
        audio_input = self.input.audio
        if audio_input is None:
            return

        async for frame in audio_input:
            if self._activity is not None:
                self._activity.push_audio(frame)

    @utils.log_exceptions(logger=logger)
    async def _forward_video_task(self) -> None:
        video_input = self.input.video
        if video_input is None:
            return

        async for frame in video_input:
            if self._activity is not None:
                if self._video_sampler is not None and not self._video_sampler(frame, self):
                    continue  # ignore this frame

                self._activity.push_video(frame)

    def _set_user_away_timer(self) -> None:
        self._cancel_user_away_timer()
        if self._opts.user_away_timeout is None:
            return

        if _RunningTasks.get(self):
            # a tool in flight will speak when it lands; the window restarts then (#6883)
            return

        if (
            (room_io := self._room_io)
            and room_io.subscribed_fut
            and not room_io.subscribed_fut.done()
        ):
            # skip the timer before user join the room
            return

        self._user_away_timer = self._loop.call_later(
            self._opts.user_away_timeout, self._update_user_state, "away"
        )

    def _cancel_user_away_timer(self) -> None:
        if self._user_away_timer is not None:
            self._user_away_timer.cancel()
            self._user_away_timer = None

    def _on_aec_warmup_expired(self) -> None:
        if self._aec_warmup_remaining > 0 and not self._closing:
            logger.debug("aec warmup expired, re-enabling interruptions")

        self._aec_warmup_remaining = 0.0
        if self._aec_warmup_timer is not None:
            self._aec_warmup_timer.cancel()
            self._aec_warmup_timer = None

    def _on_room_io_participant_linked(self, participant: rtc.RemoteParticipant) -> None:
        if (span := self._session_span) is not None and span.is_recording():
            span.add_event("participant_linked", trace_utils.participant_attributes(participant))
            if participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
                # join keys with the telephony trace; only the end user's number is PII
                span.set_attributes(
                    {
                        (
                            trace_types.ATTR_SIP_PHONE_NUMBER
                            if key == "sip.phoneNumber"
                            else trace_types.ATTR_SIP_PREFIX + key.removeprefix("sip.")
                        ): value
                        for key, value in participant.attributes.items()
                        if key.startswith("sip.")
                    }
                )

        if self._aec_warmup_duration_explicit:
            return

        is_outbound_sip = (
            participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
            and not participant.attributes.get(_SIP_RULE_ID_ATTR)
        )
        self._opts.aec_warmup_duration = None if is_outbound_sip else _DEFAULT_AEC_WARMUP_DURATION
        self._aec_warmup_remaining = self._opts.aec_warmup_duration or 0.0

        if is_outbound_sip and self._aec_warmup_timer is not None:
            self._aec_warmup_timer.cancel()
            self._aec_warmup_timer = None

    def _record_loop_stall(self, duration: float, *, timestamp_ns: int) -> None:
        """Mark a blocked event loop on the agent_session span: one event per stall plus a
        running count / total / max, so a session with stalls is findable from its list entry."""
        span = self._session_span
        if span is None or not span.is_recording():
            return
        self._loop_stall_count += 1
        self._loop_stall_total += duration
        self._loop_stall_max = max(self._loop_stall_max, duration)
        span.add_event(
            loop_monitor.SPAN_NAME,
            {trace_types.ATTR_BLOCKING_DURATION: duration},
            timestamp=timestamp_ns,
        )
        span.set_attributes(
            {
                trace_types.ATTR_BLOCKING_COUNT: self._loop_stall_count,
                trace_types.ATTR_BLOCKING_TOTAL_DURATION: self._loop_stall_total,
                trace_types.ATTR_BLOCKING_MAX_DURATION: self._loop_stall_max,
            }
        )

    def _add_session_event(
        self, name: str, attributes: dict[str, Any], *, timestamp: float | None = None
    ) -> None:
        """Timestamped marker on the agent_session span (state changes, room events)."""
        if (span := self._session_span) is None or not span.is_recording():
            return
        span.add_event(
            name,
            attributes,
            timestamp=int(timestamp * 1_000_000_000) if timestamp is not None else None,
        )

    def _update_agent_state(
        self,
        state: AgentState,
        *,
        otel_context: otel_context.Context | None = None,
        start_time: float | None = None,
    ) -> None:
        if self._agent_state == state:
            return

        start_time_ns = int(start_time * 1_000_000_000) if start_time else None

        if state == "speaking":
            self._llm_error_counts = 0
            self._tts_error_counts = 0

            if self._agent_speaking_span is None:
                self._agent_speaking_span = tracer.start_span(
                    "agent_speaking", context=otel_context, start_time=start_time_ns
                )

                if self._room_io:
                    _set_participant_attributes(
                        self._agent_speaking_span, self._room_io.room.local_participant
                    )
                # self._agent_speaking_span.set_attribute(trace_types.ATTR_START_TIME, time.time())
        elif self._agent_speaking_span is not None:
            # self._agent_speaking_span.set_attribute(trace_types.ATTR_END_TIME, time.time())
            self._agent_speaking_span.end()
            self._agent_speaking_span = None

        # aec warmup: start a one-shot wall-clock timer on the first speaking turn
        if (
            state == "speaking"
            and self._aec_warmup_remaining > 0
            and self._aec_warmup_timer is None
            and self._output.audio_enabled
            and self._output.audio is not None
        ):
            self._aec_warmup_timer = self._loop.call_later(
                self._aec_warmup_remaining, self._on_aec_warmup_expired
            )
            logger.debug(
                "aec warmup active, disabling interruptions for %.2fs",
                self._aec_warmup_remaining,
            )

        if state == "listening" and self._user_state == "listening":
            self._set_user_away_timer()
        else:
            self._cancel_user_away_timer()

        old_state = self._agent_state
        self._agent_state = state
        self._add_session_event(
            "agent_state_changed",
            {trace_types.ATTR_OLD_STATE: old_state, trace_types.ATTR_NEW_STATE: state},
        )
        self.emit(
            "agent_state_changed",
            AgentStateChangedEvent(old_state=old_state, new_state=state),
        )

    def _update_user_state(
        self, state: UserState, *, last_speaking_time: float | None = None
    ) -> None:
        # pinned to "speaking" while a `claim_user_turn` is active; voice
        # transitions are recoverable from `_user_silence_event` on release
        if self._user_turn_claims > 0 and state != "speaking":
            return

        if self._user_state == state:
            return

        last_speaking_time_ns = (
            int(last_speaking_time * 1_000_000_000) if last_speaking_time else None
        )

        if state == "speaking" and self._user_speaking_span is None:
            self._user_speaking_span = tracer.start_span(
                "user_speaking", start_time=last_speaking_time_ns
            )

            if self._room_io and self._room_io.linked_participant:
                _set_participant_attributes(
                    self._user_speaking_span, self._room_io.linked_participant
                )

            # self._user_speaking_span.set_attribute(trace_types.ATTR_START_TIME, time.time())
        elif self._user_speaking_span is not None:
            # end_time = last_speaking_time or time.time()
            # self._user_speaking_span.set_attribute(trace_types.ATTR_END_TIME, end_time)
            self._user_speaking_span.end(end_time=last_speaking_time_ns)
            self._user_speaking_span = None

        if state == "listening" and self._agent_state == "listening":
            self._set_user_away_timer()
        else:
            self._cancel_user_away_timer()

        old_state = self._user_state
        self._user_state = state
        self._add_session_event(
            "user_state_changed",
            {trace_types.ATTR_OLD_STATE: old_state, trace_types.ATTR_NEW_STATE: state},
            timestamp=last_speaking_time,
        )
        self.emit(
            "user_state_changed",
            UserStateChangedEvent(
                old_state=old_state,
                new_state=state,
                created_at=last_speaking_time or time.time(),
            ),
        )

    def _on_audio_enabled_changed(self, enabled: bool) -> None:
        """End user speaking state when audio is disabled by default."""
        if not enabled and self._user_state == "speaking":
            if self._activity is not None:
                self._activity.on_end_of_speech(None)
            else:
                self._update_user_state("listening")

    def _user_input_transcribed(self, ev: UserInputTranscribedEvent) -> None:
        if ev.transcript:
            # a transcript means stt recovered; reset its error tolerance
            self._stt_error_counts = 0

        if ev.is_final and self.user_state != "speaking":
            if self.user_state == "away":
                # reset user state from away to listening in case VAD has a miss detection
                self._update_user_state("listening")
            elif self.user_state == "listening" and self._agent_state == "listening":
                # VAD may have missed speech; STT still saw activity, so refresh away timeout
                self._set_user_away_timer()

        self.emit("user_input_transcribed", ev)

    def _conversation_item_added(self, message: llm.ChatMessage) -> None:
        self._chat_ctx.insert(message)
        if text := message.raw_text_content:
            logger.debug(
                "conversation_item_added",
                extra={"role": message.role, "lk.pii.text": text},
            )
        self.emit("conversation_item_added", ConversationItemAddedEvent(item=message))

    def _tool_items_added(self, items: Sequence[llm.FunctionCall | llm.FunctionCallOutput]) -> None:
        self._chat_ctx.insert(items)

    def _tool_execution_updated(self, ev: ToolExecutionUpdatedEvent) -> None:
        if (
            isinstance(ev.update, ToolCallEnded)
            and self._user_state == "listening"
            and self._agent_state == "listening"
        ):
            # tools hold the window off, so the last one to land restarts it (#6883)
            self._set_user_away_timer()

        self.emit("tool_execution_updated", ev)

    def _config_update_added(self, item: llm.AgentConfigUpdate) -> None:
        self._chat_ctx.insert(item)

    # move them to the end to avoid shadowing the same named modules for mypy
    @property
    def _text_only(self) -> bool:
        """True when running under a text simulation: the session uses no audio
        I/O and no audio models (STT/TTS/VAD)."""
        from ..job import current_simulation
        from ..simulation import SimulationMode

        sim = current_simulation()
        return sim is not None and sim.simulation_mode == SimulationMode.SIMULATION_MODE_TEXT

    @property
    def stt(self) -> stt.STT | None:
        return self._stt

    @property
    def llm(self) -> llm.LLM | llm.RealtimeModel | None:
        return self._llm

    @property
    def tts(self) -> tts.TTS | None:
        return self._tts

    @property
    def vad(self) -> vad.VAD | None:
        return self._vad

    @property
    def interruption_detection(self) -> NotGivenOr[Literal["adaptive", "vad"]]:
        return self._interruption_detection

    # -- User changed input/output streams/sinks --

    def _on_video_input_changed(self) -> None:
        if not self._started:
            return

        if self._forward_video_atask is not None:
            self._forward_video_atask.cancel()

        self._forward_video_atask = asyncio.create_task(
            self._forward_video_task(), name="_forward_video_task"
        )

    def _on_audio_input_changed(self) -> None:
        if not self._started:
            return

        if self._forward_audio_atask is not None:
            self._forward_audio_atask.cancel()

        self._forward_audio_atask = asyncio.create_task(
            self._forward_audio_task(), name="_forward_audio_task"
        )

    def _on_video_output_changed(self) -> None:
        pass

    def _on_audio_output_changed(self) -> None:
        if (
            self._started
            and self._opts.interruption["resume_false_interruption"]
            and (audio_output := self.output.audio)
            and not audio_output.can_pause
        ):
            logger.warning(
                "resume_false_interruption is enabled, but the audio output does not support pause, ignored",
                extra={"audio_output": audio_output.label},
            )

    def _on_text_output_changed(self) -> None:
        pass

    # ---

    async def __aenter__(self) -> AgentSession:
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        await self.aclose()

class AgentStateChangedEvent(BaseModel):
    type: Literal["agent_state_changed"] = "agent_state_changed"
    old_state: AgentState
    new_state: AgentState
    created_at: float = Field(default_factory=time.time)

class AgentTask(Agent, Generic[TaskResult_T]):
    def __init__(
        self,
        *,
        instructions: str | Instructions,
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        tools: list[llm.Tool | llm.Toolset] | None = None,
        stt: NotGivenOr[stt.STT | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        turn_handling: NotGivenOr[TurnHandlingOptions] = NOT_GIVEN,
        llm: NotGivenOr[llm.LLM | llm.RealtimeModel | None] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | None] = NOT_GIVEN,
        preserve_function_call_history: bool = False,
        # deprecated
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
        min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        mcp_servers: NotGivenOr[list[mcp.MCPServer] | None] = NOT_GIVEN,
    ) -> None:
        tools = tools or []
        turn_handling = (
            _migrate_turn_handling(
                turn_detection=turn_detection,
                allow_interruptions=allow_interruptions,
                min_endpointing_delay=min_endpointing_delay,
                max_endpointing_delay=max_endpointing_delay,
            )
            if not is_given(turn_handling)
            else turn_handling
        )
        super().__init__(
            instructions=instructions,
            chat_ctx=chat_ctx,
            tools=tools,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            mcp_servers=mcp_servers,
            turn_handling=turn_handling,
        )

        self.__started = False
        self.__fut = asyncio.Future[TaskResult_T]()
        self.__inactive_ev = asyncio.Event()
        self.__inactive_ev.set()  # set when the agent is not awaited or activity is closed
        self._preserve_function_call_history = preserve_function_call_history

        self._old_agent: Agent | None = None

    def done(self) -> bool:
        return self.__fut.done()

    def cancel(self) -> None:
        if self._activity:
            self._activity.interrupt(force=True)
        if self.__fut.done():
            return
        self.complete(ToolError(f"AgentTask {self.id} is cancelled"))

    def complete(self, result: TaskResult_T | Exception) -> None:
        if self.__fut.done():
            raise RuntimeError(f"{self.__class__.__name__} is already done")

        if isinstance(result, Exception):
            self.__fut.set_exception(result)
        else:
            self.__fut.set_result(result)

        self.__fut.exception()  # silence exc not retrieved warnings

        from .agent_activity import _SpeechHandleContextVar

        speech_handle = _SpeechHandleContextVar.get(None)

        if speech_handle:
            speech_handle._maybe_run_final_output = result

        # if not self.__inline_mode:
        #    session._close_soon(reason=CloseReason.TASK_COMPLETED, drain=True)

    async def __await_impl(self) -> TaskResult_T:
        if self.__started:
            raise RuntimeError(f"{self.__class__.__name__} is not re-entrant, await only once")

        self.__started = True

        current_task = asyncio.current_task()
        if current_task is None:
            raise RuntimeError(
                f"{self.__class__.__name__} must be executed inside an async context"
            )

        task_info = _get_activity_task_info(current_task)
        if not task_info or not task_info.inline_task:
            raise RuntimeError(
                f"{self.__class__.__name__} should only be awaited inside tool_functions or the on_enter/on_exit methods of an Agent"  # noqa: E501
            )

        # imported at call time: agent_activity imports Agent at module scope, so the reverse can't
        from .agent_activity import _AgentActivityContextVar, _SpeechHandleContextVar

        speech_handle = _SpeechHandleContextVar.get(None)
        old_activity = _AgentActivityContextVar.get()
        old_agent = old_activity.agent
        session = old_activity.session
        self._old_agent = old_agent

        def _handle_task_done(_: asyncio.Task[Any]) -> None:
            if self.__fut.done():
                return

            # if the asyncio.Task running the InlineTask completes before the InlineTask itself, log
            # an error and attempt to recover by terminating the InlineTask.
            logger.error(
                f"The asyncio.Task finished before {self.__class__.__name__} was completed."
            )

            self.complete(
                RuntimeError(
                    f"The asyncio.Task finished before {self.__class__.__name__} was completed."
                )
            )

        blocked_tasks = [current_task]
        if (
            old_activity._on_enter_task
            and not old_activity._on_enter_task.done()
            and current_task is not old_activity._on_enter_task
        ):
            blocked_tasks.append(old_activity._on_enter_task)

        async with old_activity._inline_task_slot(
            speech_handle=speech_handle, blocked_tasks=blocked_tasks
        ):
            current_task.add_done_callback(_handle_task_done)

            if (
                task_info.function_call
                and isinstance(old_activity.llm, RealtimeModel)
                and not old_activity.llm.capabilities.manual_function_calls
            ):
                logger.error(
                    f"Realtime model '{old_activity.llm.label}' does not support resuming function calls from chat context, "
                    "using AgentTask inside a function tool may have unexpected behavior."
                )

            # TODO(theomonnom): could the RunResult watcher & the blocked_tasks share the same logic?
            self.__inactive_ev.clear()
            suspended_handles: list[SpeechHandle | asyncio.Future[Any]] = []
            pending_on_enter_task: asyncio.Task[None] | None = None
            try:
                # use wait_on_enter=False to avoid deadlock: on_enter may spawn nested
                # AgentTasks that require user input, but session.run() can't return until
                # all watched handles complete — creating a circular wait.
                await session._update_activity(
                    self,
                    previous_activity="pause",
                    blocked_tasks=blocked_tasks,
                    wait_on_enter=False,
                )

                if not self._activity and not self.done():
                    self.complete(
                        ToolError(
                            f"activity doesn't start for {self.id}, likely due to session closing"
                        )
                    )

                run_state = session._global_run_state

                if self._activity and (on_enter_task := self._activity._on_enter_task):
                    if run_state and not run_state.done():
                        # watch the on_enter task as a guard so RunResult won't complete
                        # before on_enter has registered its own speech handles
                        run_state._watch_handle(on_enter_task)
                        pending_on_enter_task = on_enter_task
                    else:
                        # no active run to guard — just wait for on_enter directly
                        await asyncio.shield(on_enter_task)

                # now unwatch the parent speech handle and blocked tasks that belong to the
                # old activity — they can't complete while this AgentTask is running, and
                # keeping them watched would block RunResult from completing. A foreground
                # hold waiting on this task is in the same position, so its guard suspends too.
                if run_state and not run_state.done():
                    if speech_handle and run_state._unwatch_handle(speech_handle):
                        suspended_handles.append(speech_handle)
                    for blocked in [*blocked_tasks, *session._foreground_guards]:
                        if run_state._unwatch_handle(blocked):
                            suspended_handles.append(blocked)
                    if suspended_handles:
                        run_state._mark_done_if_needed(None)
            # asyncio.CancelledError derives from BaseException, not Exception
            except BaseException:
                self.__inactive_ev.set()
                raise

            try:
                return await asyncio.shield(self.__fut)

            finally:
                # run_state could have changed after self.__fut
                run_state = session._global_run_state

                # re-watch the suspended handles so the resumed parent activity
                # is tracked by the current RunResult again
                if run_state and not run_state.done():
                    for handle in suspended_handles:
                        run_state._watch_handle(handle)

                if pending_on_enter_task:
                    try:
                        await asyncio.shield(pending_on_enter_task)
                    except BaseException:
                        logger.exception("error in on_enter task of agent %s", self.id)

                if session._closing and self._activity is None:
                    # the activity never started (session closing), skip the handoff;
                    # the close path owns the previous activity
                    pass
                elif session.current_agent != self:
                    logger.warning(
                        f"{self.__class__.__name__} completed, but the agent has changed in the meantime. "
                        "Ignoring handoff to the previous agent, likely due to `AgentSession.update_agent` being invoked."
                    )
                    await old_activity.aclose()
                else:
                    merged_chat_ctx = old_agent.chat_ctx.merge(
                        self.chat_ctx,
                        exclude_function_call=not self._preserve_function_call_history,
                        exclude_instructions=True,
                    )
                    # set the chat_ctx directly, `session._update_activity` will sync it to the rt_session if needed
                    old_agent._chat_ctx.items[:] = merged_chat_ctx.items

                    await session._update_activity(
                        old_agent, new_activity="resume", wait_on_enter=False
                    )
                self.__inactive_ev.set()

    def __await__(self) -> Generator[None, None, TaskResult_T]:
        return self.__await_impl().__await__()

    async def _wait_for_inactive(self) -> None:
        await self.__inactive_ev.wait()

class AudioRecognition:
    def __init__(
        self,
        session: AgentSession,
        *,
        hooks: RecognitionHooks,
        endpointing: BaseEndpointing,
        stt: io.STTNode | None,
        vad: vad.VAD | None,
        interruption_detection: inference.AdaptiveInterruptionDetector | None,
        turn_detection: TurnDetectionMode | None,
        stt_model: str | None = None,
        stt_provider: str | None = None,
        stt_aligned_transcript: bool = False,
    ) -> None:
        self._session = session
        self._hooks = hooks
        self._audio_input_atask: asyncio.Task[None] | None = None
        self._commit_user_turn_atask: asyncio.Task[None] | None = None
        self._stt_consumer_atask: asyncio.Task[None] | None = None
        self._vad_atask: asyncio.Task[None] | None = None
        self._end_of_turn_task: asyncio.Task[None] | None = None
        self._endpointing: BaseEndpointing = endpointing
        self._turn_detector = turn_detection if not isinstance(turn_detection, str) else None
        self._stt = stt
        self._vad = vad
        self._stt_model = stt_model
        self._stt_provider = stt_provider
        self._stt_aligned_transcript = stt_aligned_transcript
        self._turn_detection_mode = turn_detection if isinstance(turn_detection, str) else None
        self._vad_base_turn_detection = self._turn_detection_mode in ("vad", None)
        self._user_turn_committed = False  # true if user turn ended but EOU task not done

        self._sample_rate: int | None = None

        # set on END_OF_SPEECH, cleared on START_OF_SPEECH; _speaking is its inverse.
        # exposed as an event so _wait_for_inactive can level-wait on it
        self._user_silence_ev = asyncio.Event()
        self._user_silence_ev.set()

        self._last_final_transcript_time: float | None = None
        self._last_speaking_time: float | None = None
        self._speech_start_time: float | None = None

        # used for manual commit_user_turn
        self._final_transcript_received = asyncio.Event()
        self._final_transcript_confidence: list[float] = []
        self._audio_transcript = ""
        self._audio_interim_transcript = ""
        # used for STTs that support preflight mode, so it could start preemptive generation earlier
        self._audio_preflight_transcript = ""
        self._last_language: LanguageCode | None = None

        self._stt_pipeline: _STTPipeline | None = None
        self._vad_ch: aio.Chan[rtc.AudioFrame] | None = None
        self._vad_stream: VADStream | None = None

        self._tasks: set[asyncio.Task[Any]] = set()

        # region: adaptive interruption detection
        self._interruption_atask: asyncio.Task[None] | None = None
        self._interruption_detection = interruption_detection
        self._interruption_ch: aio.Chan[inference.InterruptionDataFrameType] | None = None
        self._transcript_buffer: deque[SpeechEvent] = deque()
        self._transcript_gate_active: bool = False
        self._interruption_enabled: bool = interruption_detection is not None and vad is not None
        # Tracks active audio playout, independently of the generation lifecycle.
        self._agent_speaking: bool = False
        self._agent_speech_started_at: float | None = None
        self._active_vad_speech_started_at: float | None = None
        # turn-scoped backchannel-over-agent verdict from adaptive interruption, consumed and reset at end of turn
        self._overlap_in_current_turn: bool = False
        self._turn_backchannel_over_agent: bool = False
        # an overlap is open right now, awaiting a verdict; several can occur within one turn
        self._overlap_open: bool = False

        _backchannel_boundary: float | tuple[float, float] | None = (
            session.options.interruption.get("backchannel_boundary")
        )
        self._backchannel_boundary: tuple[float, float] | None = (
            (_backchannel_boundary, _backchannel_boundary)
            if isinstance(_backchannel_boundary, int | float)
            else _backchannel_boundary
        )
        if self._backchannel_boundary and (
            len(self._backchannel_boundary) != 2 or any(x < 0.0 for x in self._backchannel_boundary)
        ):
            raise ValueError("backchannel_boundary must be a tuple of two non-negative floats")
        self._backchannel_boundary_timer: asyncio.TimerHandle | None = None
        self._backchannel_boundary_callback: Callable[[], None] | None = None
        # endregion

        self._user_turn_span: trace.Span | None = None
        self._user_turn_start: float | None = None
        # eou_wait: one span per user turn, from the last speech anchor to the turn decision
        self._eou_wait_span: trace.Span | None = None
        self._eou_wait_started_at_ns: int | None = None
        self._eou_wait_rearms: int = 0
        self._eou_wait_not_committed: int = 0
        # eou_wait spans this user turn went through that ended with the user resuming
        self._user_turn_resumes: int = 0
        # latest timestamp recorded inside the wait; the span must not end before it
        self._eou_wait_floor_ns: int | None = None
        self._eou_detection_span: trace.Span | None = None
        self._stt_request_ids: list[str] = []
        self._closing = asyncio.Event()
        self.__stt_context: BaseModel | None = None

        self._vad_speech_started: bool = False

        self._transcription_timeout_handle: asyncio.TimerHandle | None = None
        self._turn_speech_duration: float = 0.0
        self._turn_transcript_received: bool = False

        # user turn limit tracking — accumulates across turns until agent speaks
        self._turn_tracker = _UserTurnTracker()
        self._word_tokenizer = tokenize.basic.WordTokenizer()

        # streaming audio turn detection
        self._turn_detector_stream: _StreamingTurnDetectorStream | None = None
        self._turn_detector_prediction_fut: asyncio.Future[TurnDetectionEvent] | None = None
        self._turn_detector_flushed: bool = False
        self._turn_detector_late_prediction_warned: bool = False
        self._last_emitted_prediction: TurnDetectionEvent | None = None

    def _update_options(
        self,
        *,
        endpointing: NotGivenOr[BaseEndpointing] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        # deprecated
        min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
        max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        if is_given(endpointing):
            self._endpointing = endpointing

        if is_given(turn_detection):
            self._update_turn_detector(
                turn_detection if not isinstance(turn_detection, str) else None
            )

            mode = turn_detection if isinstance(turn_detection, str) else None
            if self._turn_detection_mode != mode:
                previous_mode = self._turn_detection_mode
                self._turn_detection_mode = mode
                self._vad_base_turn_detection = self._turn_detection_mode in ("vad", None)

                if self._turn_detection_mode == "manual" or previous_mode == "manual":
                    if self._end_of_turn_task:
                        if not self._end_of_turn_task.done():
                            self._end_of_turn_task.cancel()
                    self._end_of_turn_task = None
                    # the pending decision is abandoned with the mode; the user turn stays open
                    self._end_eou_wait_span("dropped")
                    self._user_turn_committed = False
                    if self._turn_detector_stream is not None:
                        self._turn_detector_stream.cancel_inference()
                    self._turn_detector_prediction_fut = None

    def _update_last_language(self, language: LanguageCode, transcript: str) -> None:
        if not language or language.language in _NON_SPECIFIC_LANGUAGE_CODES:
            return

        if not self._last_language or len(transcript) > MIN_LANGUAGE_DETECTION_LENGTH:
            self._last_language = language

    @property
    def _input_started_at(self) -> float | None:
        return self._stt_pipeline.input_started_at if self._stt_pipeline is not None else None

    def _start(
        self,
        *,
        stt_pipeline: _STTPipeline | None = None,
        turn_detector_stream: _StreamingTurnDetectorStream | None = None,
    ) -> None:
        self._update_stt(self._stt, pipeline=stt_pipeline)
        self._update_vad(self._vad)
        self._update_interruption_detection(self._interruption_detection)
        if isinstance(self._turn_detector, _StreamingTurnDetector) or self._turn_detector is None:
            self._update_turn_detector(self._turn_detector, stream=turn_detector_stream)

    def _stop(self) -> None:
        self._update_stt(None)
        self._update_vad(None)
        self._update_turn_detector(None)
        self._update_interruption_detection(None)

    @property
    def stt_context(self) -> BaseModel | None:
        """Live speaker metadata from the STT stream.

        STT plugins set ``RecognizeStream.context`` during recognition.
        The framework copies it here so it's accessible even after the stream
        is replaced (e.g. during agent handoff).
        """
        return self.__stt_context

    @stt_context.setter
    def stt_context(self, value: BaseModel | None) -> None:
        self.__stt_context = value

    def llm_instructions(self) -> str | None:
        """Speaker context formatted as LLM instructions.

        Returns ``stt_context.to_instructions()`` if the context implements
        :class:`SpeakerContext`, otherwise ``None``.
        """
        ctx = self.__stt_context
        if ctx is not None and isinstance(ctx, stt.SpeakerContext):
            result = ctx.to_instructions()
            return result if result else None
        return None

    @property
    def _adaptive_interruption_active(self) -> bool:
        return (
            self._interruption_enabled
            and self._interruption_ch is not None
            and not self._interruption_ch.closed
        )

    # region: boundary for adaptive interruption detection

    @property
    def _backchannel_boundary_active(self) -> bool:
        return self._backchannel_boundary_timer is not None

    def _on_backchannel_boundary_done(self) -> None:
        self._backchannel_boundary_timer = None
        cb, self._backchannel_boundary_callback = (
            self._backchannel_boundary_callback,
            None,
        )
        if cb is not None:
            cb()

    def _cancel_backchannel_boundary(self) -> None:
        if self._backchannel_boundary_timer is not None:
            self._backchannel_boundary_timer.cancel()
            self._backchannel_boundary_timer = None
        self._backchannel_boundary_callback = None

    # endregion

    def _on_start_of_agent_speech(self, started_at: float) -> None:
        """Mark the start of active agent speech.

        This lifecycle follows audible playout, not the generation. Resuming paused playout
        starts a new active-speech interval.
        """
        self._transcript_gate_active = False
        self._agent_speaking = True
        self._agent_speech_started_at = started_at
        self._endpointing.on_start_of_agent_speech(started_at=started_at)

        # reset user turn tracker when agent starts speaking
        self._turn_tracker = _UserTurnTracker()

        if self._backchannel_boundary and (start_cooldown := self._backchannel_boundary[0]) > 0:
            self._cancel_backchannel_boundary()
            self._backchannel_boundary_timer = asyncio.get_running_loop().call_later(
                start_cooldown, self._on_backchannel_boundary_done
            )

        if self._adaptive_interruption_active:
            self._interruption_ch.send_nowait(_AgentSpeechStartedSentinel())  # type: ignore[union-attr]

        if self._speaking:
            self._on_start_of_overlap_speech(
                started_at=started_at,
                user_speaking_span=self._session._user_speaking_span,
            )

    def _on_end_of_agent_speech(self, *, ended_at: float) -> None:
        """Mark the end of active agent speech.

        This can occur while the generation remains active, such as when playout is paused.
        """
        self._cancel_backchannel_boundary()
        agent_was_speaking = self._agent_speaking

        if agent_was_speaking:
            self._endpointing.on_end_of_agent_speech(ended_at=ended_at)
        # Replayed STT events must observe the post-playout state.
        self._agent_speaking = False

        if not self._adaptive_interruption_active:
            self._flush_held_transcripts()
            self._overlap_open = False
            self._agent_speech_started_at = None
            return

        if agent_was_speaking:
            # close any unresolved overlap before resetting the detector
            self._on_end_of_overlap_speech(ended_at=ended_at, agent_ended=True)

        self._interruption_ch.send_nowait(_AgentSpeechEndedSentinel())  # type: ignore[union-attr]

        self._overlap_open = False

        self._flush_held_transcripts(
            resolved_at=ended_at,
            vad_speech_started_at=self._active_vad_speech_started_at,
        )

        self._agent_speech_started_at = None

    def _on_start_of_speech(
        self,
        started_at: float,
        speech_duration: float = 0.0,
        user_speaking_span: trace.Span | None = None,
        *,
        skip_adaptive_interruption: bool = False,
    ) -> None:
        self._endpointing.on_start_of_speech(
            started_at=started_at, overlapping=self._agent_speaking
        )
        # every speech onset clears the prior backchannel verdict; an overlap re-derives it below
        self._turn_backchannel_over_agent = False
        if not self._agent_speaking:
            self._overlap_in_current_turn = False

        if not skip_adaptive_interruption:
            self._on_start_of_overlap_speech(
                started_at=started_at,
                speech_duration=speech_duration,
                user_speaking_span=user_speaking_span,
            )

    def _on_start_of_overlap_speech(
        self,
        started_at: float,
        speech_duration: float = 0.0,
        user_speaking_span: trace.Span | None = None,
    ) -> None:
        if not self._adaptive_interruption_active or not self._agent_speaking:
            return

        start_boundary = self._backchannel_boundary[0] if self._backchannel_boundary else 0.0
        started_in_boundary = (
            start_boundary > 0
            and self._agent_speech_started_at is not None
            and started_at <= self._agent_speech_started_at + start_boundary
        )
        if (
            self._hooks.interruption_by_audio_activity_enabled
            or self._backchannel_boundary_active
            or started_in_boundary
        ):
            # VAD/STT owns the rest of this agent-speech interval because this
            # overlap's transcript can arrive after VAD EOS.
            self._hooks.interruption_by_audio_activity_enabled = True
            self._flush_held_transcripts()
            return

        # overlap over agent speech started this turn; gates verdict acceptance below
        self._overlap_in_current_turn = True
        self._overlap_open = True
        self._transcript_gate_active = True
        self._interruption_ch.send_nowait(  # type: ignore[union-attr]
            _OverlapSpeechStartedSentinel(
                speech_duration=speech_duration,
                user_speaking_span=user_speaking_span,
                started_at=started_at,
            )
        )

    def _on_end_of_speech(
        self,
        ended_at: float,
        user_speaking_span: trace.Span | None = None,
        interruption: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        if self._speaking:
            self._endpointing.on_end_of_speech(
                ended_at=ended_at,
                interruption=interruption,
            )

        self._on_end_of_overlap_speech(ended_at=ended_at, user_speaking_span=user_speaking_span)

    def _on_end_of_overlap_speech(
        self,
        ended_at: float,
        user_speaking_span: trace.Span | None = None,
        agent_ended: bool = False,
    ) -> None:
        """End interruption inference when agent is speaking and overlap speech ends.

        agent_ended is True when the overlap is force-ended because the agent finished
        speaking (the user may still be talking), in which case the synthesized verdict
        is inconclusive and must not be treated as a confirmed backchannel.
        """
        # the overlap ends once, on the first of: a verdict, the user stopping, the agent
        # stopping, or a teardown — so a call can arrive with it already closed
        if not self._adaptive_interruption_active or not self._overlap_open:
            return
        self._overlap_open = False

        # Only set is_interruption=false if not already set (avoid overwriting true from interruption detection)
        if user_speaking_span and user_speaking_span.is_recording():
            if isinstance(user_speaking_span, ReadableSpan):
                if (
                    user_speaking_span.attributes
                    and user_speaking_span.attributes.get(trace_types.ATTR_IS_INTERRUPTION) is None
                ):
                    user_speaking_span.set_attribute(trace_types.ATTR_IS_INTERRUPTION, "false")
            else:
                user_speaking_span.set_attribute(trace_types.ATTR_IS_INTERRUPTION, "false")

        self._interruption_ch.send_nowait(  # type: ignore[union-attr]
            _OverlapSpeechEndedSentinel(ended_at=ended_at or time.time(), agent_ended=agent_ended)
        )

    @property
    def _speaking(self) -> bool:
        return not self._user_silence_ev.is_set()

    @_speaking.setter
    def _speaking(self, value: bool) -> None:
        if value:
            self._user_silence_ev.clear()
        else:
            self._user_silence_ev.set()

    async def _wait_for_user_silence(self) -> None:
        if self._user_silence_ev.is_set():
            return
        await self._user_silence_ev.wait()

    def _trim_held_transcripts(
        self,
        *,
        resolved_at: float,
        vad_speech_started_at: float | None,
    ) -> None:
        end_boundary = self._backchannel_boundary[1] if self._backchannel_boundary else 0.0
        trim_start = resolved_at - end_boundary
        if vad_speech_started_at is not None:
            trim_start = min(trim_start, vad_speech_started_at)
        if self._agent_speech_started_at is not None:
            trim_start = max(trim_start, self._agent_speech_started_at)

        while self._transcript_buffer:
            event = self._transcript_buffer[0]
            # Known speech timing takes precedence; arrival time is the fallback.
            if event.speech_end_time is not None:
                should_trim = event.speech_end_time < trim_start and (
                    self._agent_speech_started_at is None
                    or self._agent_speech_started_at < event.speech_end_time
                )
            else:
                should_trim = event.created_at < trim_start

            if not should_trim:
                # Keep the provider-ordered suffix. Events have no stable utterance ID,
                # so filtering later events can separate SOS, transcripts, and EOS.
                break
            self._transcript_buffer.popleft()

    def _flush_held_transcripts(
        self,
        *,
        resolved_at: float | None = None,
        vad_speech_started_at: float | None = None,
    ) -> None:
        """Stop holding transcripts and emit the retained events in provider order."""
        gate_was_active = self._transcript_gate_active
        if gate_was_active or self._transcript_buffer:
            logger.trace(
                "flushing held transcripts",
                extra={
                    "event_count": len(self._transcript_buffer),
                    "gate_was_active": gate_was_active,
                    "resolved_at": resolved_at,
                    "vad_speech_started_at": vad_speech_started_at,
                },
            )

        self._transcript_gate_active = False
        if resolved_at is not None:
            self._trim_held_transcripts(
                resolved_at=resolved_at,
                vad_speech_started_at=vad_speech_started_at,
            )

        if not self._transcript_buffer:
            return

        events_to_emit = list(self._transcript_buffer)
        self._transcript_buffer.clear()
        for ev in events_to_emit:
            logger.trace("re-emitting held STT event", extra={"event": ev.type})
            self._process_stt_event(ev)

    def _set_interruption_enabled(self, enabled: bool) -> None:
        self._interruption_enabled = enabled
        if not enabled:
            self._flush_held_transcripts()

    def _push_audio(
        self, frame: rtc.AudioFrame, *, stt_frame: rtc.AudioFrame | None = None
    ) -> None:
        """Forward an audio frame to STT, VAD, AMD and the interruption detector.

        When ``stt_frame`` is provided, it is sent to the STT pipeline in place of
        ``frame`` (e.g. a silence substitute during AEC warmup or uninterruptible
        speech). VAD, AMD and the interruption channel always receive ``frame``.
        """
        self._sample_rate = frame.sample_rate
        if self._stt_pipeline is not None:
            # stamp the wall-clock anchor on the first frame to reach the pipeline
            if self._stt_pipeline.input_started_at is None:
                self._stt_pipeline.input_started_at = time.time() - frame.duration
            self._stt_pipeline.audio_ch.send_nowait(stt_frame if stt_frame is not None else frame)

        if self._vad_ch is not None:
            self._vad_ch.send_nowait(frame)

        if self._session.amd is not None:
            self._session.amd.push_audio(frame)

        if self._interruption_ch is not None:
            self._interruption_ch.send_nowait(frame)

        if self._turn_detector_stream is not None:
            self._turn_detector_stream.push_audio(frame)

    async def _aclose(self) -> None:
        self._closing.set()
        try:
            if self._commit_user_turn_atask is not None:
                result = (
                    await asyncio.gather(self._commit_user_turn_atask, return_exceptions=True)
                )[0]
                if isinstance(result, Exception):
                    logger.warning(
                        "error while committing the final user turn on close: %s",
                        type(result).__name__,
                    )

            if self._stt_pipeline is not None:
                await self._stt_pipeline.aclose()
                self._stt_pipeline = None

            await aio.cancel_and_wait(*self._tasks)

            if self._stt_consumer_atask is not None:
                await aio.cancel_and_wait(self._stt_consumer_atask)

            if self._vad_atask is not None:
                await aio.cancel_and_wait(self._vad_atask)

            if self._interruption_atask is not None:
                await aio.cancel_and_wait(self._interruption_atask)

            if self._end_of_turn_task is not None:
                result = (await asyncio.gather(self._end_of_turn_task, return_exceptions=True))[0]
                if isinstance(result, Exception):
                    logger.warning(
                        "error while completing the final user turn on close: %s",
                        type(result).__name__,
                    )

            if self._turn_detector_stream is not None:
                await self._turn_detector_stream.aclose()
                self._turn_detector_stream = None
            self._turn_detector_prediction_fut = None

            if self._backchannel_boundary_timer is not None:
                self._backchannel_boundary_timer.cancel()
                self._backchannel_boundary_timer = None
                self._backchannel_boundary_callback = None
        finally:
            self._cancel_transcription_timeout()
            # EOU normally ends this span, but teardown cancels EOU before a
            # pending speech segment necessarily produces a transcript.
            self._end_user_turn_span()

    def _update_stt(
        self,
        stt: io.STTNode | None,
        *,
        pipeline: _STTPipeline | None = None,
        model: NotGivenOr[str | None] = NOT_GIVEN,
        provider: NotGivenOr[str | None] = NOT_GIVEN,
        aligned_transcript: NotGivenOr[bool] = NOT_GIVEN,
        reset_context: bool = False,
    ) -> None:
        self._stt = stt
        # model/provider drive the user_turn span attributes; swapping to a different STT must
        # refresh them (they default to unchanged for same-STT resets like _clear_user_turn)
        if is_given(model):
            self._stt_model = model
        if is_given(provider):
            self._stt_provider = provider
        if is_given(aligned_transcript):
            self._stt_aligned_transcript = aligned_transcript
        # speaker metadata belongs to the old stream; drop it so a new STT starts clean
        if reset_context:
            self.stt_context = None
        if pipeline is None and stt is not None:
            pipeline = _STTPipeline(stt, is_closing=self._session._is_closing)
        elif pipeline is not None and stt is not None:
            # reused pipeline: rebind to this activity's node so a recreation
            # after an error doesn't call into the previous (torn-down) agent
            pipeline._rebind_node(stt)

        if pipeline is not None:
            self._stt_consumer_atask = asyncio.create_task(
                self._stt_consumer(
                    event_ch=pipeline.event_ch,
                    old_pipeline=self._stt_pipeline,
                    old_consumer=self._stt_consumer_atask,
                )
            )
            self._stt_pipeline = pipeline
            # reset interruption handling related state
            self._transcript_buffer.clear()
        else:
            self._cancel_transcription_timeout()

            if self._stt_consumer_atask is not None:
                task = asyncio.create_task(aio.cancel_and_wait(self._stt_consumer_atask))
                task.add_done_callback(lambda _: self._tasks.discard(task))
                self._tasks.add(task)
                self._stt_consumer_atask = None

            if self._stt_pipeline is not None:
                task = asyncio.create_task(self._stt_pipeline.aclose())
                task.add_done_callback(lambda _: self._tasks.discard(task))
                self._tasks.add(task)
                self._stt_pipeline = None

    def _check_vad_silence_requirement(
        self,
        detector: NotGivenOr[_TurnDetector | _StreamingTurnDetector | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> None:
        if not is_given(detector):
            detector = self._turn_detector
        # validate a candidate vad (before it's applied) when given, else the active one
        target_vad = vad if is_given(vad) else self._vad
        if not isinstance(detector, _StreamingTurnDetector) or target_vad is None:
            return
        if (current := getattr(target_vad, "min_silence_duration", None)) is None:
            return
        required = (MIN_SILENCE_DURATION_MS + 50) / 1000
        if current < required:
            raise ValueError(
                f"vad min_silence_duration={current}s is too low for the TurnDetector. "
                f"Raise the VAD's min_silence_duration to at least {required}s."
            )

    def _update_vad(self, vad: vad.VAD | None) -> None:
        self._vad = vad
        self._check_vad_silence_requirement()
        if vad:
            self._vad_stream = None
            self._vad_ch = aio.Chan[rtc.AudioFrame]()
            self._vad_atask = asyncio.create_task(
                self._vad_task(vad, self._vad_ch, self._vad_atask)
            )
        elif self._vad_atask is not None:
            task = asyncio.create_task(aio.cancel_and_wait(self._vad_atask))
            task.add_done_callback(lambda _: self._tasks.discard(task))
            self._tasks.add(task)
            self._vad_atask = None
            self._vad_ch = None
            self._vad_stream = None

        self._set_interruption_enabled(
            self._interruption_detection is not None and self._vad is not None
        )

    async def _detach_stt(self) -> _STTPipeline | None:
        """Detach the STT pipeline for handoff to another AudioRecognition.

        Returns the pipeline (pump task + channels) without stopping it.
        The caller is responsible for passing it to the new AudioRecognition
        via start(..., stt_pipeline=pipeline).
        """
        pipeline = self._stt_pipeline
        self._stt_pipeline = None

        # stop the consumer — the new AudioRecognition will start its own
        if self._stt_consumer_atask is not None:
            await aio.cancel_and_wait(self._stt_consumer_atask)
            self._stt_consumer_atask = None

        return pipeline

    def _update_interruption_detection(
        self, interruption_detection: inference.AdaptiveInterruptionDetector | None
    ) -> None:
        self._flush_held_transcripts()
        self._interruption_detection = interruption_detection
        self._overlap_open = False  # the stream it belonged to is gone either way
        if interruption_detection is not None:
            self._interruption_ch = aio.Chan[inference.InterruptionDataFrameType]()
            self._interruption_atask = asyncio.create_task(
                self._interruption_task(
                    interruption_detection, self._interruption_ch, self._interruption_atask
                )
            )
        else:
            if self._interruption_atask is not None:
                task = asyncio.create_task(aio.cancel_and_wait(self._interruption_atask))
                task.add_done_callback(lambda _: self._tasks.discard(task))
                self._tasks.add(task)
                self._interruption_atask = None
            self._interruption_ch = None
            self._cancel_backchannel_boundary()

        self._set_interruption_enabled(
            self._interruption_detection is not None and self._vad is not None
        )

    def _update_turn_detector(
        self,
        detector: _TurnDetector | _StreamingTurnDetector | None,
        *,
        stream: _StreamingTurnDetectorStream | None = None,
    ) -> None:
        """Update the turn detector and turn detector stream if possible.

        When *stream* is provided it is adopted as-is (handoff reuse) instead of
        opening a fresh stream on *detector*; the live transport stream — and its
        per-session cloud->local fallback state — survives the handoff.
        """
        self._check_vad_silence_requirement(detector)
        self._turn_detector = detector

        if (old_stream := self._turn_detector_stream) is not None and old_stream is not stream:
            task = asyncio.create_task(old_stream.aclose())
            task.add_done_callback(lambda _: self._tasks.discard(task))
            self._tasks.add(task)
        if stream is None:
            stream = detector.stream() if isinstance(detector, _StreamingTurnDetector) else None
        if self._turn_detector_stream is not stream:
            self._turn_detector_prediction_fut = None
            self._turn_detector_flushed = False
        self._turn_detector_stream = stream

    def _detach_turn_detector(self) -> _StreamingTurnDetectorStream | None:
        """Detach the turn detector stream for handoff to another AudioRecognition.

        Returns the live stream (transport run loop intact) without closing it.
        The caller passes it to the new AudioRecognition via
        ``start(..., turn_detector_stream=stream)``. The adopting recognition
        starts a fresh inference request on its next VAD event, superseding
        any request that survived the handoff.
        """
        stream, self._turn_detector_stream = self._turn_detector_stream, None
        self._turn_detector_prediction_fut = None
        return stream

    def _clear_user_turn(self) -> None:
        self._audio_transcript = ""
        self._audio_interim_transcript = ""
        self._audio_preflight_transcript = ""
        self._final_transcript_confidence = []
        self._last_final_transcript_time = None
        self._speech_start_time = None
        self._last_speaking_time = None
        self._vad_speech_started = False
        self._user_turn_committed = False
        self._last_emitted_prediction = None
        if self._turn_detector_stream is not None:
            self._turn_detector_stream.flush(reason="clear_user_turn")
            self._turn_detector_prediction_fut = None
            self._turn_detector_flushed = True

        self._turn_tracker = _UserTurnTracker()

        # end any in-progress user_turn span so the next speech starts a fresh one
        self._end_user_turn_span()
        self._stt_request_ids = []
        self._reset_transcription_timeout()

        # reset stt to clear the buffer from previous user turn
        stt = self._stt
        self._update_stt(None)
        self._update_stt(stt)

    def _commit_user_turn(
        self,
        *,
        audio_detached: bool,
        transcript_timeout: float,
        stt_flush_duration: float = 2.0,
        skip_reply: bool = False,
    ) -> asyncio.Future[str]:
        loop = asyncio.get_running_loop()
        fut: asyncio.Future[str] = loop.create_future()

        if not self._stt or self._closing.is_set():
            fut.set_result("")
            return fut

        async def _commit_user_turn() -> None:
            if self._last_final_transcript_time is None or (
                time.time() - self._last_final_transcript_time > 0.5
            ):
                # if the last final transcript is received more than 0.5s ago
                # append a silence frame to the stt to flush the buffer

                self._final_transcript_received.clear()

                # flush the stt by pushing silence
                if audio_detached and self._sample_rate:
                    silence = utils.audio.silence_frame(0.2, self._sample_rate)
                    num_frames = max(0, int(math.ceil(stt_flush_duration / silence.duration)))
                    for _ in range(num_frames):
                        self._push_audio(silence)

                # wait for the final transcript to be available
                try:
                    await asyncio.wait_for(
                        self._final_transcript_received.wait(),
                        timeout=transcript_timeout,
                    )
                except asyncio.TimeoutError:
                    if self._audio_interim_transcript:
                        logger.warning(
                            "final transcript not received after timeout",
                            extra={
                                "transcript_timeout": transcript_timeout,
                                "lk.pii.interim_transcript": self._audio_interim_transcript,
                            },
                        )

            if self._audio_interim_transcript:
                # emit interim transcript as final for frontend display
                self._hooks.on_final_transcript(
                    stt.SpeechEvent(
                        type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                        alternatives=[
                            stt.SpeechData(
                                language=LanguageCode(""), text=self._audio_interim_transcript
                            )
                        ],
                    )
                )

                # append interim transcript in case the final transcript is not ready
                self._audio_transcript = (
                    f"{self._audio_transcript} {self._audio_interim_transcript}".strip()
                )

            transcript = self._audio_transcript
            self._audio_interim_transcript = ""
            chat_ctx = self._hooks.retrieve_chat_ctx().copy()
            self._run_eou_detection(
                chat_ctx,
                skip_reply=skip_reply,
                trigger="manual",
            )
            self._user_turn_committed = True
            if not fut.done():
                fut.set_result(transcript)

        def _on_task_done(task: asyncio.Task[None]) -> None:
            if fut.done():
                return
            if task.cancelled():
                fut.cancel()
            elif exc := task.exception():
                fut.set_exception(exc)

        if self._commit_user_turn_atask is not None:
            self._commit_user_turn_atask.cancel()

        self._commit_user_turn_atask = asyncio.create_task(_commit_user_turn())
        self._commit_user_turn_atask.add_done_callback(_on_task_done)
        return fut

    @property
    def _current_transcript(self) -> str:
        """
        Transcript for this turn, including interim transcript if available.
        """
        if self._audio_interim_transcript:
            return self._audio_transcript + " " + self._audio_interim_transcript
        return self._audio_transcript

    async def _on_stt_event(self, ev: stt.SpeechEvent) -> None:
        if (
            ev.speech_end_time is None
            and self._stt_aligned_transcript
            and ev.alternatives
            and ev.alternatives[0].end_time > 0
            and self._input_started_at is not None
        ):
            speech_end_time = self._input_started_at + ev.alternatives[0].end_time
            if speech_end_time <= ev.created_at:
                ev = replace(ev, speech_end_time=speech_end_time)

        # Collect provider-known STT ids for this user turn. The actual attribute
        # is written once when the user_turn span ends (see _on_end_of_turn), to
        # avoid ordering issues with span creation.
        if ev.request_id and ev.request_id not in self._stt_request_ids:
            self._stt_request_ids.append(ev.request_id)

        if (
            self._turn_detection_mode == "manual"
            and self._user_turn_committed
            and (
                self._end_of_turn_task is None
                or self._end_of_turn_task.done()
                or ev.type == stt.SpeechEventType.INTERIM_TRANSCRIPT
            )
        ):
            # ignore transcript for manual turn detection when user turn already committed
            # and EOU task is done or this is an interim transcript
            return

        if (
            ev.type == stt.SpeechEventType.FINAL_TRANSCRIPT
            and ev.alternatives
            and ev.alternatives[0].text
        ):
            self._mark_turn_transcribed()
            # whenever there is a final transcript when the gate is off
            # fall back to vad interruption for the rest of the turn:
            # - late STT transcript during backchannel boundary
            # - late STT transcript when no overlap speech has been detected
            if self._agent_speaking and not self._transcript_gate_active:
                self._hooks.interruption_by_audio_activity_enabled = True

        if ev.type != stt.SpeechEventType.RECOGNITION_USAGE and self._transcript_gate_active:
            logger.trace("holding STT event during agent speech", extra={"event": ev.type})
            self._transcript_buffer.append(ev)
            return

        self._process_stt_event(ev)

    def _process_stt_event(self, ev: stt.SpeechEvent) -> None:
        has_stt_end_time = bool(
            len(ev.alternatives) > 0
            and ev.alternatives[0].end_time > 0
            and self._input_started_at is not None
        )
        now = time.time()
        stt_last_speaking_time = (
            min(ev.alternatives[0].end_time + self._input_started_at, now)
            if has_stt_end_time and self._input_started_at is not None
            else now
        )
        # Prefer the provider's speaking time when there is no VAD anchor to beat:
        # no VAD at all, or the VAD missed this segment. In STT turn detection the
        # provider owns the turn boundary, so its timestamp wins there too
        # one issue to note: without `end_time` the estimate collapses to `now`
        # which would report a ~0 transcription_delay for STT mode.
        use_stt_speaking_time = (
            self._vad is None
            or self._last_speaking_time is None
            or (self._turn_detection_mode == "stt" and has_stt_end_time)
        )
        if ev.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
            transcript = ev.alternatives[0].text
            language = ev.alternatives[0].language
            confidence = ev.alternatives[0].confidence

            self._update_last_language(language, transcript)

            self._final_transcript_received.set()
            if not transcript:
                return

            self._hooks.on_final_transcript(
                ev,
                speaking=self._speaking
                if (self._vad is not None) or self._turn_detection_mode == "stt"
                else None,
            )
            if self._session.amd is not None:
                self._session.amd._on_transcript(transcript)

            extra: dict[str, Any] = {
                "lk.pii.user_transcript": transcript,
                "language": self._last_language,
            }
            if self._last_speaking_time:
                extra["transcript_delay"] = time.time() - self._last_speaking_time
            logger.debug("received user transcript", extra=extra)

            self._last_final_transcript_time = time.time()
            self._audio_transcript += f" {transcript}"
            self._audio_transcript = self._audio_transcript.lstrip()
            self._final_transcript_confidence.append(confidence)
            transcript_changed = self._audio_transcript != self._audio_preflight_transcript
            self._audio_interim_transcript = ""
            self._audio_preflight_transcript = ""

            if use_stt_speaking_time:
                self._last_speaking_time = stt_last_speaking_time

            # check user turn limit after accumulating transcript
            self._check_user_turn_limit(transcript)

            if self._vad_base_turn_detection or self._user_turn_committed:
                if transcript_changed:
                    self._hooks.on_preemptive_generation(
                        _PreemptiveGenerationInfo(
                            new_transcript=self._audio_transcript,
                            transcript_confidence=(
                                sum(self._final_transcript_confidence)
                                / len(self._final_transcript_confidence)
                                if self._final_transcript_confidence
                                else 0
                            ),
                            started_speaking_at=self._speech_start_time,
                        )
                    )

                if not self._speaking:
                    chat_ctx = self._hooks.retrieve_chat_ctx().copy()
                    self._run_eou_detection(
                        chat_ctx,
                        trigger="stt",
                    )

        elif ev.type == stt.SpeechEventType.PREFLIGHT_TRANSCRIPT:
            self._hooks.on_interim_transcript(
                ev,
                speaking=self._speaking
                if (self._vad is not None) or self._turn_detection_mode == "stt"
                else None,
            )
            transcript = ev.alternatives[0].text
            language = ev.alternatives[0].language
            confidence = ev.alternatives[0].confidence

            self._update_last_language(language, transcript)

            if not transcript:
                return

            logger.debug(
                "received user preflight transcript",
                extra={"lk.pii.user_transcript": transcript, "language": self._last_language},
            )

            # still need to increment it as it's used for turn detection,
            self._last_final_transcript_time = time.time()
            # preflight transcript includes all pre-committed transcripts (including final transcript from the previous STT run)
            self._audio_preflight_transcript = (self._audio_transcript + " " + transcript).lstrip()
            self._audio_interim_transcript = transcript

            if use_stt_speaking_time:
                self._last_speaking_time = stt_last_speaking_time

            if self._turn_detection_mode != "manual" or self._user_turn_committed:
                confidence_vals = list(self._final_transcript_confidence) + [confidence]
                self._hooks.on_preemptive_generation(
                    _PreemptiveGenerationInfo(
                        new_transcript=self._audio_preflight_transcript,
                        transcript_confidence=sum(confidence_vals) / len(confidence_vals),
                        started_speaking_at=self._speech_start_time,
                    )
                )

        elif ev.type == stt.SpeechEventType.INTERIM_TRANSCRIPT:
            self._hooks.on_interim_transcript(
                ev,
                speaking=self._speaking
                if (self._vad is not None) or self._turn_detection_mode == "stt"
                else None,
            )
            self._audio_interim_transcript = ev.alternatives[0].text

        elif ev.type == stt.SpeechEventType.END_OF_SPEECH and self._turn_detection_mode == "stt":
            with tracer.use_span(self._ensure_user_turn_span()):
                self._hooks.on_end_of_speech(None)

            # STT EOT changes user state from speaking to listening without updating VAD internal states
            # VAD EOS will also skip updating user state from listening (STT enforced) to listening (VAD detected)
            # and user state won't be updated until a new VAD SOS is received
            # reset VAD so that incorrect end of turn from STT can be corrected by VAD interruption
            # if user is still speaking (an immediate VAD SOS will interrupt the agent)
            if self._vad:
                if self._vad_speech_started:
                    if self._vad_stream is not None:
                        self._vad_stream.flush()
                    else:
                        self._update_vad(self._vad)

                    logger.warning(
                        "stt end of speech received while vad is still in a speech segment, "
                        "flushing vad",
                        extra={
                            "vad_speech_start_time": self._speech_start_time,
                            "flushed": self._vad_stream is not None,
                        },
                    )

            self._speaking = False
            self._user_turn_committed = True

            # always use STT speaking time since turn detection mode is set to STT. we would want
            # alignment here since _last_speaking_time is used for turn detection timing
            if ev.speech_end_time is not None:
                # clamped like the other anchors: a provider clock running ahead would
                # otherwise push the anchor into the future and extend `extra_sleep`,
                # delaying the turn commit by the skew
                self._last_speaking_time = min(ev.speech_end_time, now)
            else:
                # use an implied version computed based on either word timestamps or current time
                self._last_speaking_time = stt_last_speaking_time

            chat_ctx = self._hooks.retrieve_chat_ctx().copy()
            self._run_eou_detection(
                chat_ctx,
                trigger="stt",
            )

        elif ev.type == stt.SpeechEventType.START_OF_SPEECH and self._turn_detection_mode == "stt":
            # If the plugin provided a server onset timestamp, use it;
            # otherwise fall back to message arrival time.
            if self._speech_start_time is None:
                self._speech_start_time = ev.speech_start_time or time.time()
            self._end_eou_wait_span("user_resumed", end_time=ev.speech_start_time or now)

            with tracer.use_span(self._ensure_user_turn_span(start_time=self._speech_start_time)):
                self._hooks.on_start_of_speech(None, speech_start_time=self._speech_start_time)

            self._speaking = True
            self._last_speaking_time = stt_last_speaking_time

            if self._end_of_turn_task is not None:
                self._end_of_turn_task.cancel()

    @utils.log_exceptions(logger=logger)
    async def _on_vad_event(self, ev: vad.VADEvent) -> None:
        if ev.type == vad.VADEventType.START_OF_SPEECH:
            speech_start_time = time.time() - ev.speech_duration - ev.inference_duration
            self._active_vad_speech_started_at = speech_start_time
            if not self._vad_speech_started:
                self._speech_start_time = speech_start_time
                self._vad_speech_started = True

            self._cancel_transcription_timeout()
            self._end_eou_wait_span("user_resumed", end_time=speech_start_time)

            with tracer.use_span(self._ensure_user_turn_span(start_time=speech_start_time)):
                self._hooks.on_start_of_speech(ev, speech_start_time=speech_start_time)

            self._speaking = True

            if self._turn_detector_stream is not None:
                self._turn_detector_stream.cancel_inference()
            self._turn_detector_prediction_fut = None
            self._turn_detector_flushed = False

            if self._end_of_turn_task is not None:
                self._end_of_turn_task.cancel()

            if self._session.amd is not None:
                self._session.amd._on_user_speech_started()

        elif ev.type == vad.VADEventType.INFERENCE_DONE:
            self._hooks.on_vad_inference_done(ev)

            # for metrics, get the "earliest" signal of speech as possible
            if ev.raw_accumulated_speech > 0.0:
                self._last_speaking_time = time.time()

                if self._speech_start_time is None:
                    self._speech_start_time = time.time() - ev.raw_accumulated_speech
                if self._speaking and self._turn_detector_prediction_fut is not None:
                    if self._turn_detector_stream is not None:
                        self._turn_detector_stream.cancel_inference()
                    self._turn_detector_prediction_fut = None

            if ev.raw_accumulated_silence >= MIN_SILENCE_DURATION_MS / 1000 and self._speaking:
                if (
                    self._turn_detector_stream is not None
                    and self._turn_detector_prediction_fut is None
                ):
                    self._turn_detector_prediction_fut = self._turn_detector_stream.predict()

        elif ev.type == vad.VADEventType.END_OF_SPEECH:
            vad_speech_started = self._vad_speech_started
            with tracer.use_span(self._ensure_user_turn_span()):
                self._hooks.on_end_of_speech(ev)

            self._active_vad_speech_started_at = None
            self._vad_speech_started = False
            self._speaking = False
            speech_end_time = time.time() - ev.silence_duration - ev.inference_duration
            self._last_speaking_time = speech_end_time

            # A committed turn clears _vad_speech_started before its late VAD EOS arrives.
            if self._stt_pipeline is not None and vad_speech_started:
                self._arm_transcription_timeout(
                    ev.speech_duration,
                    delay=ev.silence_duration + ev.inference_duration,
                )

            if self._vad_base_turn_detection or (
                self._turn_detection_mode == "stt" and self._user_turn_committed
            ):
                chat_ctx = self._hooks.retrieve_chat_ctx().copy()
                self._run_eou_detection(chat_ctx, trigger="vad")

            if self._session.amd is not None:
                self._session.amd._on_user_speech_ended(ev.silence_duration)

    def _on_overlap_speech_event(self, ev: inference.OverlappingSpeechEvent) -> None:
        # every verdict is terminal for its overlap, including one the cooldown then ignores
        self._overlap_open = False

        if self._backchannel_boundary_active and not ev.is_interruption:
            logger.trace(
                "ignoring backchannel event during backchannel boundary cooldown, falling back to vad"
            )
            return

        if ev.is_interruption and self._transcript_gate_active:
            self._flush_held_transcripts(
                resolved_at=ev.detected_at,
                vad_speech_started_at=ev.overlap_started_at,
            )
        elif self._transcript_gate_active:
            self._trim_held_transcripts(
                resolved_at=ev.detected_at,
                vad_speech_started_at=self._active_vad_speech_started_at,
            )

        # only honor the verdict while this turn's overlap is unresolved so a late verdict
        # can't leak into the next turn; an interruption supersedes a prior backchannel
        if self._overlap_in_current_turn and not ev.agent_ended:
            self._turn_backchannel_over_agent = not ev.is_interruption
            # clear the backchannel audio, but only between segments — else we'd clip a real turn
            if not ev.is_interruption and not self._speaking:
                self._hooks.on_backchannel_confirmed()

    def _on_missing_eot_prediction(self) -> None:
        if self._turn_detector_flushed:
            if not self._turn_detector_late_prediction_warned:
                self._turn_detector_late_prediction_warned = True
                logger.warning(
                    "transcript arrives after turn has been committed. consider raising `min_delay` in the "
                    "endpointing options to accommodate a slow stt. subsequent "
                    "occurrences will log at debug level.",
                )
            else:
                logger.debug("stt transcript arrived after a turn flush, skipping eot prediction")
        else:
            logger.debug("no eot inference request in flight, skipping eot prediction")

    def _run_eou_detection(
        self,
        chat_ctx: llm.ChatContext,
        *,
        trigger: Literal["vad", "stt", "manual"],
        skip_reply: bool = False,
    ) -> None:
        if self._stt and not self._audio_transcript and self._turn_detection_mode != "manual":
            # stt enabled but no transcript yet
            return

        chat_ctx = chat_ctx.copy()
        if self._audio_transcript:
            chat_ctx.add_message(role="user", content=self._audio_transcript)

        turn_detector = (
            (
                self._turn_detector_stream
                if isinstance(self._turn_detector, _StreamingTurnDetector)
                else self._turn_detector
            )
            if self._turn_detection_mode != "manual"
            and (self._audio_transcript or isinstance(self._turn_detector, _StreamingTurnDetector))
            else None  # disable EOU model if manual turn detection enabled
        )

        @utils.log_exceptions(logger=logger)
        async def _bounce_eou_task(
            last_speaking_time: float | None = None,
            last_final_transcript_time: float | None = None,
            speech_start_time: float | None = None,
        ) -> None:
            endpointing_delay = self._endpointing.min_delay
            # a turn created here (no VAD/start-of-speech opened it) starts at the earliest
            # anchor known, so the eou_wait child back-dated to last_speaking_time fits inside
            anchors = [t for t in (speech_start_time, last_speaking_time) if t is not None]
            user_turn_span = self._ensure_user_turn_span(
                start_time=min(anchors) if anchors else None
            )
            eou_wait_span = self._ensure_eou_wait_span(
                user_turn_span,
                trigger=trigger,
                last_speaking_time=last_speaking_time,
                endpointing_delay=endpointing_delay,
            )

            end_of_turn_probability: float | None = None
            unlikely_threshold: float | None = None
            backchannel_threshold: float | None = None

            if turn_detector is not None:
                if not await turn_detector.supports_language(self._last_language):
                    logger.info("Turn detector does not support language %s", self._last_language)
                else:
                    # ended explicitly: _end_eou_wait_span closes it early if the user resumes
                    with (
                        tracer.use_span(eou_wait_span),
                        tracer.start_as_current_span(
                            "eou_detection", end_on_exit=False
                        ) as eou_detection_span,
                        self._track_eou_detection_span(eou_detection_span),
                    ):
                        from_cache = False
                        prediction_event: TurnDetectionEvent | None = None
                        if isinstance(turn_detector, _StreamingTurnDetectorStream):
                            fut = self._turn_detector_prediction_fut
                            if fut is None:
                                if trigger == "stt":
                                    self._on_missing_eot_prediction()
                            else:
                                from_cache = fut.done()
                                prediction_timeout = turn_detector.prediction_timeout
                                done, _ = await asyncio.wait([fut], timeout=prediction_timeout)
                                if fut in done and not fut.cancelled():
                                    prediction_event = fut.result()
                                    end_of_turn_probability = (
                                        prediction_event.end_of_turn_probability
                                    )
                                    unlikely_threshold = await turn_detector.unlikely_threshold(
                                        self._last_language
                                    )
                                    backchannel_threshold = (
                                        await turn_detector.backchannel_threshold(
                                            self._last_language
                                        )
                                    )
                                else:
                                    logger.warning(
                                        "eot prediction timed out, committing without a prediction",
                                        extra={"timeout": prediction_timeout},
                                    )
                                    turn_detector.cancel_inference(timed_out=True)
                                    self._turn_detector_prediction_fut = None
                        else:
                            try:
                                end_of_turn_probability = await turn_detector.predict_end_of_turn(
                                    chat_ctx,
                                )
                                unlikely_threshold = await turn_detector.unlikely_threshold(
                                    self._last_language
                                )
                            except Exception:
                                logger.exception("Error predicting end of turn")

                        if (
                            end_of_turn_probability is not None
                            and unlikely_threshold is not None
                            and end_of_turn_probability < unlikely_threshold
                        ):
                            endpointing_delay = self._endpointing.max_delay

                        eou_span_attributes: dict[str, Any] = {
                            trace_types.ATTR_CHAT_CTX: json.dumps(
                                llm.ChatContext(chat_ctx.items[-_EOU_MAX_HISTORY_TURNS:])
                                .copy(
                                    exclude_function_call=True,
                                    exclude_instructions=True,
                                    exclude_empty_message=True,
                                    exclude_handoff=True,
                                    exclude_config_update=True,
                                )
                                .to_dict(
                                    exclude_audio=True,
                                    exclude_image=True,
                                    exclude_timestamp=True,
                                    exclude_metrics=True,
                                )
                            ),
                            trace_types.ATTR_EOU_DELAY: endpointing_delay,
                            trace_types.ATTR_EOU_LANGUAGE: self._last_language or "",
                            trace_types.ATTR_EOU_SOURCE: trigger,
                            trace_types.ATTR_EOU_FROM_CACHE: from_cache,
                        }
                        if end_of_turn_probability is not None:
                            eou_span_attributes[trace_types.ATTR_EOU_PROBABILITY] = (
                                end_of_turn_probability
                            )
                        if unlikely_threshold is not None:
                            eou_span_attributes[trace_types.ATTR_EOU_UNLIKELY_THRESHOLD] = (
                                unlikely_threshold
                            )
                        eou_detection_span.set_attributes(eou_span_attributes)
                        logger.debug(
                            "eot prediction",
                            extra={
                                "probability": end_of_turn_probability,
                                "unlikely_threshold": unlikely_threshold,
                                "endpointing_delay": endpointing_delay,
                                "language": self._last_language or "",
                                "trigger": trigger,
                                "from_cache": from_cache,
                            },
                        )

                        if (
                            end_of_turn_probability is not None
                            and unlikely_threshold is not None
                            and (
                                prediction_event is None
                                or prediction_event is not self._last_emitted_prediction
                            )
                        ):
                            self._last_emitted_prediction = prediction_event
                            inference_duration = (
                                prediction_event.inference_duration
                                if prediction_event is not None
                                and prediction_event.inference_duration is not None
                                else 0.0
                            )
                            # end of speech -> prediction receive time
                            delay = (
                                time.time() - last_speaking_time
                                if last_speaking_time is not None
                                else 0.0
                            )
                            self._hooks.on_eot_prediction(
                                EotPredictionEvent(
                                    probability=end_of_turn_probability,
                                    threshold=unlikely_threshold,
                                    inference_duration=inference_duration,
                                    delay=delay,
                                )
                            )
                            # surface the backchannel opportunity whenever it clears its
                            # threshold, regardless of end-of-turn; AgentActivity decides
                            # whether to acknowledge mid-turn or let it lead the reply
                            backchannel_probability = (
                                prediction_event.backchannel_probability
                                if prediction_event is not None
                                else None
                            )
                            if (
                                backchannel_probability is not None
                                and backchannel_threshold is not None
                                and backchannel_probability >= backchannel_threshold
                            ):
                                self._hooks.on_agent_backchannel_opportunity(
                                    _AgentBackchannelOpportunityEvent(
                                        probability=backchannel_probability,
                                        threshold=backchannel_threshold,
                                        end_of_turn_probability=end_of_turn_probability,
                                        end_of_turn_threshold=unlikely_threshold,
                                        language=self._last_language,
                                    )
                                )
                        if (
                            prediction_event is not None
                            and prediction_event.detection_delay is not None
                        ):
                            eou_detection_span.set_attribute(
                                trace_types.ATTR_EOU_DETECTION_DELAY,
                                prediction_event.detection_delay,
                            )

            if eou_wait_span.is_recording():  # the wait may have ended with resumed speech
                eou_wait_span.set_attribute(trace_types.ATTR_EOU_DELAY, endpointing_delay)

            extra_sleep = endpointing_delay
            if last_speaking_time:
                extra_sleep += last_speaking_time - time.time()
            delay_completed = False
            if extra_sleep > 0:
                try:
                    await asyncio.wait_for(self._closing.wait(), timeout=extra_sleep)
                except asyncio.TimeoutError:
                    delay_completed = True
                    pass

            confidence_avg = (
                sum(self._final_transcript_confidence) / len(self._final_transcript_confidence)
                if self._final_transcript_confidence
                else 0
            )

            # sometimes, we can't calculate the metrics because VAD was unreliable or
            # the speaking anchor is stale/out-of-order (see issue #6093). in this case,
            # we just ignore the calculation, it's better than providing likely wrong values
            metrics = _compute_end_of_turn_metrics(
                speech_start_time=speech_start_time,
                last_speaking_time=last_speaking_time,
                last_final_transcript_time=last_final_transcript_time,
                now=time.time(),
            )
            end_of_turn = _EndOfTurnInfo(
                skip_reply=skip_reply,
                new_transcript=self._audio_transcript,
                transcript_confidence=confidence_avg,
                metrics=metrics,
                backchannel_over_agent=self._turn_backchannel_over_agent,
                user_turn_span=user_turn_span,
            )
            committed = self._hooks.on_end_of_turn(end_of_turn)
            if committed:
                logger.debug(
                    "user turn committed",
                    extra={
                        "last_speaking_time": last_speaking_time,
                        "last_final_transcript_time": last_final_transcript_time,
                        "speech_start_time": speech_start_time,
                        "delay_completed": delay_completed,
                        "source": trigger,
                        "end_of_turn_probability": end_of_turn_probability,
                        "unlikely_threshold": unlikely_threshold,
                    },
                )
                user_turn_span.set_attributes(
                    {
                        trace_types.ATTR_USER_TRANSCRIPT: self._audio_transcript,
                        trace_types.ATTR_TRANSCRIPT_CONFIDENCE: confidence_avg,
                        trace_types.ATTR_TRANSCRIPTION_DELAY: metrics.transcription_delay or 0,
                        trace_types.ATTR_END_OF_TURN_DELAY: metrics.end_of_turn_delay or 0,
                    }
                )
                if self._stt_request_ids:
                    user_turn_span.set_attribute(
                        trace_types.ATTR_PROVIDER_REQUEST_IDS, self._stt_request_ids
                    )
                self._end_eou_wait_span("committed")
                self._stamp_user_turn_resumes(user_turn_span)
                if not end_of_turn.user_turn_span_adopted:
                    user_turn_span.end()
                self._user_turn_span = None
                self._user_turn_start = None
                self._stt_request_ids = []
                self._reset_transcription_timeout()

                # clear the transcript if the user turn was committed
                self._audio_transcript = ""
                self._final_transcript_confidence = []
                self._last_final_transcript_time = None
                # concurrent user speech might have changed it
                # only reset if there is no new speech
                if self._last_speaking_time == last_speaking_time:
                    self._speech_start_time = None
                    self._vad_speech_started = False
                    self._last_speaking_time = None

                if self._turn_detector_stream is not None:
                    self._turn_detector_stream.flush(reason="turn committed")
                    self._turn_detector_prediction_fut = None
                    self._turn_detector_flushed = True

            elif eou_wait_span.is_recording():
                self._eou_wait_not_committed += 1

            # reset turn-scoped barge-in state once per logical turn (commit or drop)
            self._turn_backchannel_over_agent = False
            self._overlap_in_current_turn = False
            self._user_turn_committed = False

        if self._end_of_turn_task is not None:
            # TODO(theomonnom): disallow cancel if the extra sleep is done
            self._end_of_turn_task.cancel()
        # copy the last_speaking_time before awaiting (the value can change)
        self._end_of_turn_task = asyncio.create_task(
            _bounce_eou_task(
                self._last_speaking_time,
                self._last_final_transcript_time,
                self._user_turn_start,
            )
        )

    def _check_user_turn_limit(self, transcript: str) -> None:
        """Check if the user turn exceeds configured limits.
        Called when a final transcript event is received."""
        opts = self._session.options.turn_handling["user_turn_limit"]
        max_words = opts.get("max_words")
        max_duration = opts.get("max_duration")

        if max_words is None and max_duration is None:
            return

        now = time.time()
        if self._turn_tracker.started_at is None:
            self._turn_tracker.started_at = self._speech_start_time or now

        words = self._word_tokenizer.tokenize(transcript)
        self._turn_tracker.words += len(words)
        self._turn_tracker.transcript = f"{self._turn_tracker.transcript} {transcript}".strip()

        duration = now - self._turn_tracker.started_at
        time_exceeded = max_duration is not None and duration >= max_duration
        words_exceeded = max_words is not None and self._turn_tracker.words >= max_words

        if not time_exceeded and not words_exceeded:
            return

        ev = UserTurnExceededEvent(
            transcript=self._current_transcript,
            accumulated_transcript=self._turn_tracker.transcript,
            accumulated_word_count=self._turn_tracker.words,
            duration=duration,
        )
        self._hooks.on_user_turn_exceeded(ev)

    @utils.log_exceptions(logger=logger)
    async def _stt_consumer(
        self,
        event_ch: aio.Chan[stt.SpeechEvent],
        old_pipeline: _STTPipeline | None,
        old_consumer: asyncio.Task[None] | None,
    ) -> None:
        """Consume STT events from the pump. Swapped on handoff."""

        if old_pipeline is not None:
            await old_pipeline.aclose()

        if old_consumer is not None:
            await aio.cancel_and_wait(old_consumer)

        async for ev in event_ch:
            await self._on_stt_event(ev)

    @utils.log_exceptions(logger=logger)
    async def _vad_task(
        self,
        vad: vad.VAD,
        audio_input: AsyncIterable[rtc.AudioFrame],
        task: asyncio.Task[None] | None,
    ) -> None:
        if task is not None:
            await aio.cancel_and_wait(task)

        stream = vad.stream()
        self._vad_stream = stream

        @utils.log_exceptions(logger=logger)
        async def _forward() -> None:
            async for frame in audio_input:
                stream.push_frame(frame)

        forward_task = asyncio.create_task(_forward())

        try:
            async for ev in stream:
                await self._on_vad_event(ev)
        finally:
            await aio.cancel_and_wait(forward_task)
            await stream.aclose()
            if self._vad_stream is stream:
                self._vad_stream = None

            # reset the speaking state to prevent stuck user speaking state during handoff
            if self._speaking:
                with tracer.use_span(self._ensure_user_turn_span()):
                    self._hooks.on_end_of_speech(None)
                self._speaking = False
                self._vad_speech_started = False
            self._active_vad_speech_started_at = None

    @utils.log_exceptions(logger=logger)
    async def _interruption_task(
        self,
        interruption_detection: inference.AdaptiveInterruptionDetector,
        audio_input: AsyncIterable[inference.InterruptionDataFrameType],
        task: asyncio.Task[None] | None,
    ) -> None:
        if task is not None:
            await aio.cancel_and_wait(task)

        stream = interruption_detection.stream()

        @utils.log_exceptions(logger=logger)
        async def _forward() -> None:
            async for frame in audio_input:
                stream.push_frame(frame)

        forward_task = asyncio.create_task(_forward())

        try:
            async for ev in stream:
                self._hooks.on_overlap_speech(ev)
        except APIError:
            # avoid already emitted error from the stream
            return
        finally:
            await aio.cancel_and_wait(forward_task)
            await stream.aclose()
            if self._interruption_ch is audio_input:
                self._flush_held_transcripts()

    def _cancel_transcription_timeout(self) -> None:
        if (handle := getattr(self, "_transcription_timeout_handle", None)) is not None:
            handle.cancel()
            self._transcription_timeout_handle = None

    def _reset_transcription_timeout(self) -> None:
        self._cancel_transcription_timeout()
        self._turn_speech_duration = 0.0
        self._turn_transcript_received = False

    def _mark_turn_transcribed(self) -> None:
        self._turn_transcript_received = True
        self._cancel_transcription_timeout()

    def _arm_transcription_timeout(self, speech_duration: float, *, delay: float) -> None:
        timeout = self._session.options.transcription_timeout
        if timeout is None or self._turn_transcript_received:
            return

        self._turn_speech_duration += speech_duration
        self._cancel_transcription_timeout()
        remaining_timeout = max(0.0, timeout - delay)
        self._transcription_timeout_handle = asyncio.get_running_loop().call_later(
            remaining_timeout, self._on_transcription_timeout
        )

    def _on_transcription_timeout(self) -> None:
        self._transcription_timeout_handle = None
        if self._user_turn_start is None or self._turn_transcript_received:
            return

        self._hooks.on_transcription_timeout(
            speech_duration=self._turn_speech_duration, turn_start=self._user_turn_start
        )

    def _ensure_user_turn_span(self, start_time: float | None = None) -> trace.Span:
        if self._user_turn_span and self._user_turn_span.is_recording():
            return self._user_turn_span

        if start_time is None:
            start_time = time.time()
        start_time_ns = int(start_time * 1_000_000_000)
        # pinned to the session root: a turn may be created from any task (a late STT final)
        self._user_turn_span = tracer.start_span(
            "user_turn", context=self._session._root_span_context, start_time=start_time_ns
        )

        if self._user_turn_start is None:
            self._user_turn_start = start_time

        if (room_io := self._session._room_io) and room_io.linked_participant:
            _set_participant_attributes(self._user_turn_span, room_io.linked_participant)

        # add STT model/provider attributes
        if self._stt_model:
            self._user_turn_span.set_attribute(
                trace_types.ATTR_GEN_AI_REQUEST_MODEL, self._stt_model
            )
        if self._stt_provider:
            self._user_turn_span.set_attribute(
                trace_types.ATTR_GEN_AI_PROVIDER_NAME, self._stt_provider
            )

        return self._user_turn_span

    def _end_user_turn_span(self) -> None:
        # a wait still open here never reached a decision (teardown, clear_user_turn, ...)
        self._end_eou_wait_span("dropped")
        if self._user_turn_span is not None and self._user_turn_span.is_recording():
            self._stamp_user_turn_resumes(self._user_turn_span)
            self._user_turn_span.end()
        self._user_turn_span = None
        self._user_turn_start = None

    def _stamp_user_turn_resumes(self, user_turn_span: trace.Span) -> None:
        resumes, self._user_turn_resumes = self._user_turn_resumes, 0
        if resumes and user_turn_span.is_recording():
            user_turn_span.set_attribute(trace_types.ATTR_EOU_RESUME_COUNT, resumes)

    def _ensure_eou_wait_span(
        self,
        user_turn_span: trace.Span,
        *,
        trigger: str,
        last_speaking_time: float | None,
        endpointing_delay: float,
    ) -> trace.Span:
        """The turn's ``eou_wait`` span, created on the first end-of-turn trigger and back-dated
        to ``last_speaking_time``. Later triggers for the same turn (a late STT final, another
        VAD end of speech) re-arm the wait as an event rather than start a new span."""
        span = self._eou_wait_span
        if span is not None and span.is_recording():
            self._eou_wait_rearms += 1
            self._eou_wait_floor_ns = time.time_ns()
            span.add_event(
                "rearmed",
                {trace_types.ATTR_EOU_SOURCE: trigger},
                timestamp=self._eou_wait_floor_ns,
            )
            span.set_attributes(
                {
                    trace_types.ATTR_EOU_SOURCE: trigger,
                    # the delay in force now; a prediction may raise it later
                    trace_types.ATTR_EOU_DELAY: endpointing_delay,
                }
            )
            return span

        now = time.time()
        started_at = min(last_speaking_time, now) if last_speaking_time is not None else now
        # lk.eou.wait_duration is computed from this same value, so it equals the span's length
        started_at_ns = int(started_at * 1_000_000_000)
        with tracer.use_span(user_turn_span):
            span = tracer.start_span(
                "eou_wait",
                start_time=started_at_ns,
                attributes={
                    trace_types.ATTR_EOU_SOURCE: trigger,
                    trace_types.ATTR_EOU_DELAY: endpointing_delay,
                },
            )
        self._eou_wait_span = span
        self._eou_wait_started_at_ns = started_at_ns
        self._eou_wait_rearms = 0
        self._eou_wait_not_committed = 0
        self._eou_wait_floor_ns = None
        return span

    @contextlib.contextmanager
    def _track_eou_detection_span(self, span: trace.Span) -> Iterator[None]:
        """Own the ``eou_detection`` span's end so the wait can close it early."""
        self._eou_detection_span = span
        try:
            yield
        finally:
            if self._eou_detection_span is span:
                self._eou_detection_span = None
            if span.is_recording():
                self._eou_wait_floor_ns = time.time_ns()
                span.end(end_time=self._eou_wait_floor_ns)

    def _end_eou_wait_span(self, outcome: str, *, end_time: float | None = None) -> None:
        span, self._eou_wait_span = self._eou_wait_span, None
        started_at_ns, self._eou_wait_started_at_ns = self._eou_wait_started_at_ns, None
        rearms, self._eou_wait_rearms = self._eou_wait_rearms, 0
        not_committed, self._eou_wait_not_committed = self._eou_wait_not_committed, 0
        if span is None or not span.is_recording():
            return

        requested_ns = int((end_time if end_time is not None else time.time()) * 1_000_000_000)
        ended_at_ns = requested_ns
        if started_at_ns is not None:
            # resumed speech can carry a VAD timestamp from before the anchor; never negative
            ended_at_ns = max(ended_at_ns, started_at_ns)

        detection, self._eou_detection_span = self._eou_detection_span, None
        if detection is not None and detection.is_recording():
            # the detector is still running: end it first so the child stays inside the parent
            self._eou_wait_floor_ns = time.time_ns()
            detection.add_event(
                "superseded",
                {trace_types.ATTR_EOU_OUTCOME: outcome},
                timestamp=self._eou_wait_floor_ns,
            )
            detection.end(end_time=self._eou_wait_floor_ns)
        floor_ns, self._eou_wait_floor_ns = self._eou_wait_floor_ns, None
        if floor_ns is not None:
            # VAD reports a resume after the fact: never end before what the span contains
            ended_at_ns = max(ended_at_ns, floor_ns)
        if outcome == "user_resumed":
            span.add_event("user_resumed", timestamp=max(requested_ns, started_at_ns or 0))
        span.set_attributes(
            {
                trace_types.ATTR_EOU_OUTCOME: outcome,
                # same integers as the span bounds, so the attribute equals the span's length
                trace_types.ATTR_EOU_WAIT_DURATION: (
                    (ended_at_ns - started_at_ns) / 1_000_000_000
                    if started_at_ns is not None
                    else 0.0
                ),
                trace_types.ATTR_EOU_REARM_COUNT: rearms,
                trace_types.ATTR_EOU_NOT_COMMITTED_COUNT: not_committed,
            }
        )
        span.end(end_time=ended_at_ns)
        if outcome == "user_resumed":
            self._user_turn_resumes += 1

class CloseEvent(BaseModel):
    type: Literal["close"] = "close"
    error: (
        LLMError | STTError | TTSError | RealtimeModelError | InterruptionDetectionError | None
    ) = None
    reason: CloseReason
    created_at: float = Field(default_factory=time.time)

@unique
class CloseReason(str, Enum):
    ERROR = "error"
    JOB_SHUTDOWN = "job_shutdown"
    PARTICIPANT_DISCONNECTED = "participant_disconnected"
    USER_INITIATED = "user_initiated"
    TASK_COMPLETED = "task_completed"

class ConversationItemAddedEvent(BaseModel):
    type: Literal["conversation_item_added"] = "conversation_item_added"
    item: ChatMessage | AgentHandoff | _TypeDiscriminator
    created_at: float = Field(default_factory=time.time)

class ErrorEvent(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    type: Literal["error"] = "error"
    error: LLMError | STTError | TTSError | RealtimeModelError | InterruptionDetectionError | Any
    source: LLM | STT | TTS | RealtimeModel | AdaptiveInterruptionDetector | Any
    created_at: float = Field(default_factory=time.time)

    @field_serializer("source")
    def _serialize_source(self, source: Any) -> Any:
        if isinstance(source, LLM | STT | TTS | RealtimeModel | AdaptiveInterruptionDetector):
            return {"model": source.model, "provider": source.provider}
        if isinstance(source, BaseModel):
            return source.model_dump()
        return repr(source)

    @field_serializer("error")
    def _serialize_error(self, error: Any) -> Any:
        if isinstance(error, BaseModel):
            return error.model_dump()
        return repr(error)

class ExpressiveOptions(TypedDict, total=False):
    """Configuration for the expressive pipeline, passed as ``AgentSession(expressive=...)``
    or ``Agent(expressive=...)`` (the agent value overrides the session's).

    Controls how TTS markup instructions are injected into the LLM when expressive is
    enabled. All keys are optional; common shapes:

    - ``{"speech_steering": {...}}`` — steer delivery and non-verbal sounds on top of
      the provider-agnostic default instructions.
    - ``{"tts_instructions_template": "..."}`` — a fully custom prompt.
    - ``{"tts_instructions_append": "..."}`` — your own rules appended to the template.

    Any explicit template overrides the default; unset parts fall back to the
    provider-agnostic default.
    """

    speech_steering: SpeechSteeringOptions
    tts_instructions_template: Instructions | str
    tts_instructions_append: str

class FunctionToolsExecutedEvent(BaseModel):
    """Emitted after a batch of function tools finishes executing.

    ``function_calls`` and ``function_call_outputs`` are parallel lists: the
    output at a given index belongs to the call at the same index and carries
    the same ``call_id``. Every call has one output, even one whose tool raised
    ``StopResponse``; such an output asks for no reply with ``reply_required``.
    """

    type: Literal["function_tools_executed"] = "function_tools_executed"
    function_calls: list[FunctionCall]
    function_call_outputs: list[FunctionCallOutput]
    created_at: float = Field(default_factory=time.time)
    _handoff_required: bool = PrivateAttr(default=False)

    def zipped(self) -> list[tuple[FunctionCall, FunctionCallOutput]]:
        """Return calls paired with outputs by list position."""
        return list(zip(self.function_calls, self.function_call_outputs, strict=False))

    def cancel_tool_reply(self) -> None:
        for fnc_call_out in self.function_call_outputs:
            fnc_call_out.reply_required = False

    def cancel_agent_handoff(self) -> None:
        self._handoff_required = False

    @property
    def has_tool_reply(self) -> bool:
        return any(fnc_call_out.reply_required for fnc_call_out in self.function_call_outputs)

    @property
    def has_agent_handoff(self) -> bool:
        return self._handoff_required

    @model_validator(mode="after")
    def verify_lists_length(self) -> Self:
        if len(self.function_calls) != len(self.function_call_outputs):
            raise ValueError("The number of function_calls and function_call_outputs must match.")

        return self

class KeytermDetectionOptions(TypedDict, total=False):
    """Configuration for automatic keyterm detection.

    Lives under the ``keyterm_detection`` key of :class:`KeytermsOptions`. Absent or
    ``{"enabled": False}`` keeps detection off.
    """

    enabled: bool
    """Whether to run the background detector. Defaults to ``False``."""
    llm: LLM | str | None
    """LLM used for extraction. An ``LLM`` instance, or a model string (e.g.
    ``"google/gemini-3.5-flash"``) resolved via the inference gateway. Defaults to a
    built-in detection model; the agent's own LLM is not used."""
    turn_interval: int
    """Run a pass once per N user turns. Defaults to ``1``."""
    max_keyterms: int | None
    """Cap on the confirmed (applied) detected keyterms if provided. Defaults to ``None``."""
    instructions: str | None
    """Override the built-in extraction prompt."""
    timeout: float
    """Seconds a single detection pass may run before it is dropped (no keyterm change).
    Defaults to ``10.0``. Raise it if a slow detection ``llm`` needs longer."""

class KeytermsOptions(TypedDict, total=False):
    """Keyterm biasing for STTs that accept a term list.

    .. deprecated:: v2.0
        Use :class:`STTContextOptions` (``AgentSession(stt_context_options=...)``) instead;
        its ``keyterms`` and ``keyterm_detection`` keys are identical.
    """

    keyterms: list[str]
    """Static keyterms applied wherever the STT accepts a term list; never touched by detection."""
    keyterm_detection: KeytermDetectionOptions
    """LLM-based keyterm extraction, for STTs that accept a term list."""

class MetricsCollectedEvent(BaseModel):
    """Deprecated: use session_usage_updated for usage tracking.
    Per-turn latency metrics are available on ChatMessage.metrics."""

    type: Literal["metrics_collected"] = "metrics_collected"
    metrics: AgentMetrics
    created_at: float = Field(default_factory=time.time)

@dataclass
class ModelSettings:
    tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN
    """The tool choice to use when calling the LLM."""

class NonverbalOptions(TypedDict, total=False):
    """Non-verbal vocalizations the TTS may produce (sounds that aren't words).

    A sparse opt-out: omitted keys default to ON, and a category set to
    ``False`` is never advertised to the LLM — ``{"laughing": False}``
    removes laughter and nothing else. Together the keys cover every sound
    the providers offer. Each key may later widen to ``bool | <Sound>Options``
    (e.g. ``laughing: bool | LaughingOptions``) for per-sound control without
    breaking existing callers.
    """

    laughing: bool
    """laugh, chuckle, giggle — and laugh-speak delivery"""
    breathing: bool
    """audible breath, inhale, exhale"""
    sighing: bool
    crying: bool
    vocalizing: bool
    """non-lexical voiced sounds — humming a tune, sing-song or sung delivery"""
    mouth_sounds: bool
    """tsk, tongue-click, lip-smack"""
    reflex_sounds: bool
    """cough, clearing the throat, yawn"""

class RecordingOptions(TypedDict, total=False):
    """Granular control over which recording features are active.

    Recording keys default to ``True`` when not specified, so ``{"logs": False}``
    means "record everything except logs." Redaction defaults to the project setting;
    ``False`` is ignored when redaction is enabled globally for the project.

    Can be passed directly to :pymethod:`AgentSession.start(record=...)`:

    * ``record=True``  → all on (backward compatible)
    * ``record=False`` → all off (backward compatible)
    * ``record={"audio": True, "traces": False}`` → granular
    * ``record={"redaction": True}`` → enable redaction for the session
    """

    audio: bool
    """Record session audio. Defaults to ``True``."""
    traces: bool
    """Export OpenTelemetry trace spans. Defaults to ``True``."""
    logs: bool
    """Export OpenTelemetry logs. Defaults to ``True``."""
    transcript: bool
    """Upload the conversation transcript (chat history). Defaults to ``True``."""
    redaction: bool
    """Enable redaction. ``False`` does not disable project redaction."""

class RemoteSession(rtc.EventEmitter[RemoteSessionEventTypes]):
    def __init__(self, transport: SessionTransport) -> None:
        super().__init__()
        self._transport = transport
        self._started = False
        self._pending_requests: dict[str, asyncio.Future[agent_pb.SessionResponse]] = {}
        self._recv_task: asyncio.Task[None] | None = None

    @classmethod
    def from_room(cls, room: rtc.Room, agent_identity: str) -> RemoteSession:
        transport = RoomSessionTransport(room, agent_identity)
        return cls(transport)

    async def start(self) -> None:
        if self._started:
            return
        self._started = True
        await self._transport.start()
        self._recv_task = asyncio.create_task(self._recv_loop())

    async def aclose(self) -> None:
        if not self._started:
            return
        self._started = False

        for future in self._pending_requests.values():
            future.cancel()
        self._pending_requests.clear()

        if self._recv_task:
            await utils.aio.cancel_and_wait(self._recv_task)

        await self._transport.close()

    async def _recv_loop(self) -> None:
        try:
            async for msg in self._transport:
                if msg.HasField("response"):
                    self._dispatch_response(msg.response)
                elif msg.HasField("event"):
                    event_field = msg.event.WhichOneof("event")
                    if event_field:
                        self.emit(event_field, msg.event)
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.warning("error processing session message", exc_info=True)
        finally:
            for future in self._pending_requests.values():
                if not future.done():
                    future.set_exception(RuntimeError("remote session transport closed"))
            self._pending_requests.clear()

    def _dispatch_response(self, response: agent_pb.SessionResponse) -> None:
        future = self._pending_requests.pop(response.request_id, None)
        if future and not future.done():
            future.set_result(response)

    async def _send_request(
        self,
        request: agent_pb.SessionRequest,
        timeout: float = 60.0,
    ) -> agent_pb.SessionResponse:
        if self._recv_task is None:
            raise RuntimeError("remote session not started")
        if self._recv_task.done():
            raise RuntimeError("remote session transport closed")

        req_type = request.WhichOneof("request")
        future: asyncio.Future[agent_pb.SessionResponse] = asyncio.Future()
        self._pending_requests[request.request_id] = future

        try:
            msg = agent_pb.AgentSessionMessage(request=request)
            await self._transport.send_message(msg)
            resp = await asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            logger.warning(
                "remote session request timed out",
                extra={"request_id": request.request_id, "type": req_type, "timeout": timeout},
            )
            raise
        finally:
            self._pending_requests.pop(request.request_id, None)
            if future.done() and not future.cancelled():
                future.exception()

        if resp.error:
            raise RuntimeError(f"session request {req_type} failed: {resp.error}")

        return resp

    async def wait_for_ready(self, timeout: float = 5.0, retry_interval: float = 0.5) -> None:
        deadline = asyncio.get_event_loop().time() + timeout
        # a transport that never came up is the useful answer; a bare timeout
        # would say only that no pong arrived
        send_error: Exception | None = None
        while True:
            remaining = deadline - asyncio.get_event_loop().time()
            if remaining <= 0:
                if send_error is not None:
                    raise send_error
                raise TimeoutError("wait_for_ready timed out")
            req = agent_pb.SessionRequest(
                request_id=utils.shortuuid("req_"),
                ping=agent_pb.SessionRequest.Ping(),
            )
            try:
                await self._send_request(req, timeout=min(retry_interval, remaining))
                return
            except (TimeoutError, asyncio.TimeoutError):
                if asyncio.get_event_loop().time() >= deadline:
                    raise TimeoutError("wait_for_ready timed out") from None
            except Exception as e:
                if self._recv_task is None or self._recv_task.done():
                    raise
                # the transport reports that it is not up yet instead of dropping
                # the ping, and that state is exactly what this call polls
                # through. The request timeout paced the retries before; pace
                # them here, since the failure is now immediate.
                send_error = e
                if asyncio.get_event_loop().time() >= deadline:
                    raise
                await asyncio.sleep(
                    min(retry_interval, max(0.0, deadline - asyncio.get_event_loop().time()))
                )

    async def get_chat_history(self) -> agent_pb.SessionResponse.GetChatHistoryResponse:
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            get_chat_history=agent_pb.SessionRequest.GetChatHistory(),
        )
        resp = await self._send_request(req)
        return resp.get_chat_history

    async def get_agent_info(self) -> agent_pb.SessionResponse.GetAgentInfoResponse:
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            get_agent_info=agent_pb.SessionRequest.GetAgentInfo(),
        )
        resp = await self._send_request(req)
        return resp.get_agent_info

    async def get_framework_info(self) -> agent_pb.SessionResponse.GetFrameworkInfoResponse:
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            get_framework_info=agent_pb.SessionRequest.GetFrameworkInfo(),
        )
        resp = await self._send_request(req)
        return resp.get_framework_info

    async def get_session_state(self) -> agent_pb.SessionResponse.GetSessionStateResponse:
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            get_session_state=agent_pb.SessionRequest.GetSessionState(),
        )
        resp = await self._send_request(req)
        return resp.get_session_state

    async def run(
        self, text: str, timeout: float = 60.0
    ) -> agent_pb.SessionResponse.RunInputResponse:
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            run_input=agent_pb.SessionRequest.RunInput(text=text),
        )
        resp = await self._send_request(req, timeout=timeout)
        return resp.run_input

    async def update_io(
        self,
        *,
        input_audio_enabled: bool | None = None,
        input_video_enabled: bool | None = None,
        output_audio_enabled: bool | None = None,
        output_video_enabled: bool | None = None,
        output_transcription_enabled: bool | None = None,
        timeout: float = 60.0,
    ) -> agent_pb.SessionResponse.UpdateIOResponse:
        """Toggle the agent's I/O channels remotely.

        Only the channels passed (non-None) are applied; the rest are left
        untouched. Simulators use this to disable the agent's audio I/O instead
        of relying on a room attribute.
        """
        update = agent_pb.SessionRequest.UpdateIO()
        if input_audio_enabled is not None:
            update.input.audio_enabled = input_audio_enabled
        if input_video_enabled is not None:
            update.input.video_enabled = input_video_enabled
        if output_audio_enabled is not None:
            update.output.audio_enabled = output_audio_enabled
        if output_video_enabled is not None:
            update.output.video_enabled = output_video_enabled
        if output_transcription_enabled is not None:
            update.output.transcription_enabled = output_transcription_enabled

        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            update_io=update,
        )
        resp = await self._send_request(req, timeout=timeout)
        return resp.update_io

    async def finalize_simulation(
        self,
        *,
        provisional_success: bool,
        provisional_reason: str = "",
        timeout: float = 60.0,
    ) -> agent_pb.SessionResponse.FinalizeSimulationResponse:
        """Hand the agent under test the simulator's provisional verdict and return the
        agent's own verdict from its on_simulation_end callback. The response's
        ``user_verdict`` is unset when the agent has no handler (or times out) or sets
        no verdict of its own; both verdicts are reported, neither overrides the other."""
        req = agent_pb.SessionRequest(
            request_id=utils.shortuuid("req_"),
            finalize_simulation=agent_pb.SessionRequest.FinalizeSimulation(
                provisional_success=provisional_success,
                provisional_reason=provisional_reason,
            ),
        )
        resp = await self._send_request(req, timeout=timeout)
        return resp.finalize_simulation

class RunContext(Generic[Userdata_T]):
    # private ctor
    def __init__(
        self,
        *,
        session: AgentSession[Userdata_T],
        speech_handle: SpeechHandle,
        function_call: FunctionCall,
        activity: AgentActivity | None = None,
    ) -> None:
        self._activity = activity
        self._session = session
        self._speech_handle = speech_handle
        self._function_call = function_call

        self._initial_step_idx = speech_handle.num_steps - 1
        self._filler_schedulers: list[_FillerScheduler] = []

        # synthesized progress-update pairs, populated whether or not an executor is attached
        self._updates: list[tuple[FunctionCall, FunctionCallOutput]] = []

        # set/cleared by the executor around the tool's lifetime
        self._executor: _ToolExecutor | None = None
        self._first_update_fut: asyncio.Future[Any] | None = None

        # the run this call belongs to; background work that outlives it must not hold a
        # later run open
        self._run_state = session._global_run_state

    @property
    def session(self) -> AgentSession[Userdata_T]:
        return self._session

    @property
    def speech_handle(self) -> SpeechHandle:
        return self._speech_handle

    @property
    def function_call(self) -> FunctionCall:
        return self._function_call

    @property
    def userdata(self) -> Userdata_T:
        return self.session.userdata

    def disallow_interruptions(self) -> None:
        """Disable interruptions for this FunctionCall.

        Also releases an active false-interruption pause.

        Raises:
            RuntimeError: If the SpeechHandle is already interrupted.
        """
        if self._activity is not None:
            self._activity._disallow_interruptions(self.speech_handle)
        else:
            self.speech_handle.allow_interruptions = False

    async def wait_for_playout(self) -> None:
        """Waits for the speech playout corresponding to this function call step.

        Unlike `SpeechHandle.wait_for_playout`, which waits for the full
        assistant turn to complete (including all function tools),
        this method only waits for the assistant's spoken response prior running
        this tool to finish playing."""
        await self.speech_handle._wait_for_generation(step_idx=self._initial_step_idx)

    @asynccontextmanager
    async def with_filler(
        self,
        source: _FillerSource,
        *,
        delay: float = 0,
        interval: float | None = None,
        max_steps: int | None = None,
    ) -> AsyncIterator[None]:
        """Schedule filler speech while a long-running step blocks the tool.

        While the context is open, a background scheduler waits for the session to be
        continuously idle for ``delay`` seconds, then plays ``source``. With ``interval``
        set, it then sleeps that many wall-clock seconds before restarting the dwell
        wait. ``interval=None`` (default) fires at most once.

        Args:
            source: Either a string (spoken via ``session.say``), or a callable
                ``(step: int) -> SpeechHandle | str | None`` invoked at fire time with
                the iteration count. Returning ``None`` skips this fire and retries on
                the next interval; the step counter only advances when a handle is
                produced. Use ``max_steps`` to cap the total number of fires.
            delay: Continuous-idle dwell required before each fire. ``0`` = fire as
                soon as the session is next idle.
            interval: Wall-clock cooldown after each fire. ``None`` = fire at most once.
            max_steps: Maximum number of fires across the lifetime of the cm.
                ``None`` = no limit.
        """
        scheduler = _FillerScheduler(
            session=self._session,
            speech_handle=self._speech_handle,
            source=source,
            delay=delay,
            interval=interval,
            max_steps=max_steps,
        )
        self._filler_schedulers.append(scheduler)
        try:
            yield
        finally:
            await scheduler.aclose()
            self._filler_schedulers.remove(scheduler)

    @asynccontextmanager
    async def foreground(self) -> AsyncIterator[AgentActivity]:
        """Wait for idle, then hold the floor while interactive work runs.

        Use cases:

        - wrap an ``await AgentTask()`` so it doesn't race with current speech
          or another tool's queued reply
        - wrap a direct ``generate_reply`` / ``say`` for the same reason
        - group multiple interactive calls so no deferred tool reply lands between them

        On enter, drains this tool's pending deferred reply first so its speech
        plays before the floor is held — keeps chat order matching code order.
        """
        await self._drain_pending_reply()
        async with self._session._wait_for_idle_and_hold(run_state=self._run_state) as activity:
            yield activity

    async def update(
        self,
        message: str | Any,
        *,
        template: str | Callable[[UpdatePromptArgs], str] | None = None,
    ) -> None:
        """Push a progress update into the conversation.

        The first update releases control to the LLM with ``message`` as the tool's
        synthetic return; subsequent updates are coalesced into a deferred reply.
        Outside the voice path (e.g. ``execute_function_call``) updates are recorded
        on the result but no reply is fired.

        Args:
            message: Progress message; strings are wrapped by ``template``.
            template: Per-call override — either a ``str.format()`` template or a
                callable receiving ``UpdatePromptArgs``. Defaults to the executor's
                resolved ``update`` template (or the module default when standalone).
        """
        # update() is a deliberate agent action — reset any active filler dwell so a
        # pending filler doesn't race the real update to the speech queue
        for s in self._filler_schedulers:
            s.reset_dwell()

        # events carry the raw message, before the LLM-facing template wraps it
        raw_message = message if isinstance(message, str) else str(message)

        if isinstance(message, str):
            if template is None:
                if self._executor is not None:
                    template = self._executor._tool_options["update_template"]
                else:
                    from .tool_executor import UPDATE_TEMPLATE

                    template = UPDATE_TEMPLATE
            from .tool_executor import _render

            message = _render(
                template,
                {
                    "function_name": self.function_call.name,
                    "call_id": self.function_call.call_id,
                    "message": message,
                },
            )

        # first update keeps the original call_id
        update_step = len(self._updates)
        pair = self._make_update_pair(
            message, call_id_suffix=f"_update_{update_step}" if update_step > 0 else ""
        )
        self._updates.append(pair)

        if self._executor is None:
            return  # standalone — no executor, so no tool lifecycle to report

        self._session._tool_execution_updated(
            ToolExecutionUpdatedEvent(
                update=ToolCallUpdated(
                    id=pair[0].call_id,
                    call_id=self.function_call.call_id,
                    message=raw_message,
                )
            ),
        )

        assert self._first_update_fut is not None
        if not self._first_update_fut.done():
            self._first_update_fut.set_result(message)
            self._function_call.extra["__livekit_agents_tool_non_blocking"] = True
            return

        await self._executor._enqueue_reply(self, [pair[0], pair[1]])

    def _attach_executor(
        self, executor: _ToolExecutor, first_update_fut: asyncio.Future[Any]
    ) -> None:
        if self._first_update_fut is not None:
            raise ValueError("Executor already attached")
        self._executor = executor
        self._first_update_fut = first_update_fut

    def _detach_executor(self) -> None:
        self._executor = None
        self._first_update_fut = None

    async def _drain_pending_reply(self) -> None:
        """Wait for this tool's pending deferred reply to finish delivery, if any."""
        if self._executor is None:
            return
        reply_task = self._executor._reply_task
        if reply_task is None or reply_task.done():
            return
        try:
            await asyncio.shield(reply_task)
        except Exception:
            pass  # reply task's own errors aren't our concern

    def _make_update_pair(
        self, message: Any, *, call_id_suffix: str = ""
    ) -> tuple[FunctionCall, FunctionCallOutput]:
        """Synthesize a (FunctionCall, FunctionCallOutput) pair for a progress update.

        The new FunctionCall carries ``{call_id}{call_id_suffix}``; name/arguments/extra
        are copied. ``make_tool_output`` is reused so error handling matches dispatch.
        """
        from .generation import make_tool_output

        fnc_call = FunctionCall(
            call_id=f"{self.function_call.call_id}{call_id_suffix}",
            name=self.function_call.name,
            arguments=self.function_call.arguments,
            extra=dict(self.function_call.extra),
        )
        tool_output = make_tool_output(fnc_call=fnc_call, output=message, exception=None)
        return (fnc_call, tool_output.fnc_call_out)

class RunOutputOptions(TypedDict, total=False):
    """Structured-output behavior for :meth:`AgentSession.run`.

    Can be passed as a plain dict::

        sess.run(
            user_input=...,
            output_type=MyOutput,
            output_options={"max_retries": 2, "retry_instructions": "Call submit_result."},
        )

    Pass ``output_options=None`` to disable the retry behavior.
    """

    max_retries: int
    """Re-prompts when a run ends without its ``output_type``, before raising
    UnexpectedModelBehavior. Defaults to ``2``."""
    retry_instructions: str
    """Override the built-in retry prompt."""

class STTContextOptions(TypedDict, total=False):
    """Conversation-aware context for the STT.

    Can be passed as a plain dict::

        AgentSession(
            stt_context_options={
                "keyterms": ["LiveKit", "Acme Corp"],
                "keyterm_detection": {"enabled": True, "turn_interval": 1},
                "forward_chat_context": True,
            },
        )
    """

    keyterms: list[str]
    """Static keyterms applied wherever the STT accepts a term list; never touched by detection."""
    keyterm_detection: KeytermDetectionOptions
    """LLM-based keyterm extraction, for STTs that accept a term list."""
    forward_chat_context: bool
    """Forward conversation turns to STTs that consume context directly (e.g. AssemblyAI u3-pro).
    Defaults to ``True``; only STTs that advertise the ``chat_context`` capability act on it."""

class SessionUsageUpdatedEvent(BaseModel):
    type: Literal["session_usage_updated"] = "session_usage_updated"
    usage: AgentSessionUsage
    created_at: float = Field(default_factory=time.time)

class SpeechCreatedEvent(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    type: Literal["speech_created"] = "speech_created"
    user_initiated: bool
    """True if the speech was created using public methods like `say` or `generate_reply`"""
    source: Literal["say", "generate_reply"]
    """Source indicating how the speech handle was created"""
    speech_handle: SpeechHandle = Field(..., exclude=True)
    """The speech handle that was created"""
    created_at: float = Field(default_factory=time.time)

class SpeechHandle:
    SPEECH_PRIORITY_LOW = 0
    """Priority for messages that should be played after all other messages in the queue"""
    SPEECH_PRIORITY_NORMAL = 5
    """Every speech generates by the VoiceAgent defaults to this priority."""
    SPEECH_PRIORITY_HIGH = 10
    """Priority for important messages that should be played before others."""

    def __init__(
        self, *, speech_id: str, allow_interruptions: bool, input_details: InputDetails
    ) -> None:
        self._id = speech_id
        self._allow_interruptions = allow_interruptions
        self._interruption_holds = 0
        self._interruption_holds_restore = allow_interruptions
        self._input_details = input_details

        self._interrupt_fut = asyncio.Future[None]()
        self._done_fut = asyncio.Future[None]()
        self._scheduled_fut = asyncio.Future[None]()
        self._authorize_event = asyncio.Event()

        self._generations: list[asyncio.Future[None]] = []

        # internal tasks used by this generation
        self._tasks: list[asyncio.Task] = []
        self._chat_items: list[llm.ChatItem] = []
        self._num_steps = 1
        # one agent_turn span for the whole speech, however many generations (LLM steps) it
        # takes; opened by the first reply task, ended with the speech in _mark_done
        self._agent_turn_span: trace.Span | None = None
        self._agent_turn_context: otel_context.Context | None = None
        self._agent_turn_started_at: float | None = None
        self._agent_turn_agent_name: str | None = None
        self._scheduled_at: float | None = None
        self._authorized_at: float | None = None
        self._interrupt_source: InterruptionSource | None = None  # first interrupt's cause

        self._interrupt_timeout_handle: asyncio.TimerHandle | None = None

        self._item_added_callbacks: set[Callable[[llm.ChatItem], None]] = set()
        self._done_callbacks: set[Callable[[SpeechHandle], None]] = set()

        def _on_done(_: asyncio.Future[None]) -> None:
            for cb in list(self._done_callbacks):
                try:
                    cb(self)
                except Exception as e:
                    logger.warning(f"error in done_callback: {cb}", exc_info=e)

        self._done_fut.add_done_callback(_on_done)
        self._maybe_run_final_output: Any = None  # kept private
        self._error: BaseException | None = None

    @staticmethod
    def create(
        allow_interruptions: bool = True,
        input_details: InputDetails = DEFAULT_INPUT_DETAILS,
    ) -> SpeechHandle:
        return SpeechHandle(
            speech_id=utils.shortuuid("speech_"),
            allow_interruptions=allow_interruptions,
            input_details=input_details,
        )

    @property
    def num_steps(self) -> int:
        return self._num_steps

    @property
    def id(self) -> str:
        return self._id

    @property
    def input_details(self) -> InputDetails:
        return self._input_details

    @property
    def _generation_id(self) -> str:
        return f"{self._id}_{self._num_steps}"

    @property
    def _parent_generation_id(self) -> str | None:
        if self._num_steps <= 1:
            return None
        return f"{self._id}_{self._num_steps - 1}"

    @property
    def scheduled(self) -> bool:
        return self._scheduled_fut.done()

    @property
    def interrupted(self) -> bool:
        return self._interrupt_fut.done()

    @property
    def allow_interruptions(self) -> bool:
        return self._allow_interruptions

    @allow_interruptions.setter
    def allow_interruptions(self, value: bool) -> None:
        """Allow or disallow interruptions on this SpeechHandle.

        When set to False, the SpeechHandle will no longer accept any incoming
        interruption requests until re-enabled. If the handle is already
        interrupted, clearing interruptions is not allowed.

        Args:
            value (bool): True to allow interruptions, False to disallow.

        Raises:
            RuntimeError: If attempting to disable interruptions when already interrupted.
        """
        if self.interrupted and not value:
            raise RuntimeError(
                "Cannot set allow_interruptions to False, the SpeechHandle is already interrupted"
            )

        self._allow_interruptions = value

    def _hold_interruptions(self) -> None:
        """Disallow interruptions until every hold taken here is released.

        Counted rather than set, because the holders of one speech are not serialised
        against each other: the inline tasks awaited from a turn's parallel tool calls run
        one at a time, and a hold released between them would let the user turns of one
        task's sub-conversation interrupt the speech the rest are still anchored to. An
        interrupted handle can no longer disallow interruptions, so those tasks could then
        never run. The first holder owns the value the last one restores.
        """
        if self._interruption_holds == 0:
            self._interruption_holds_restore = self._allow_interruptions
            self.allow_interruptions = False

        self._interruption_holds += 1

    def _release_interruptions(self) -> None:
        self._interruption_holds -= 1
        if self._interruption_holds == 0:
            # a forced interrupt lands regardless of the hold, and leaves nothing to restore
            with contextlib.suppress(RuntimeError):
                self.allow_interruptions = self._interruption_holds_restore

    @property
    def chat_items(self) -> list[llm.ChatItem]:
        return self._chat_items

    def done(self) -> bool:
        return self._done_fut.done()

    def exception(self) -> BaseException | None:
        """Return the error that caused this speech to fail, if any.

        Awaiting a SpeechHandle never raises; call this method after the handle
        is done to check whether the generation failed (e.g. ``llm.RealtimeError``
        when a realtime reply timed out).

        Raises:
            asyncio.InvalidStateError: If the speech is not done yet.

        Returns:
            BaseException | None: The error the generation failed with, or None.
        """
        if not self._done_fut.done():
            raise asyncio.InvalidStateError("SpeechHandle is not done yet")

        return self._error

    def interrupt(
        self, *, force: bool = False, source: InterruptionSource = "programmatic"
    ) -> SpeechHandle:
        """Interrupt the current speech generation.

        Args:
            force: Interrupt even if this speech disallows interruptions.
            source: Why, for the ``agent_turn`` trace (see ``InterruptionSource``). The first
                interruption's cause is the one recorded.

        Raises:
            RuntimeError: If this speech handle is still running and does not allow
                interruptions.

        Returns:
            SpeechHandle: The same speech handle that was interrupted.
        """
        if self.interrupted or self.done():
            # already cancelled or finished: nothing to interrupt, and protection is moot
            return self

        if not force and not self._allow_interruptions:
            raise RuntimeError("This generation handle does not allow interruptions")

        self._interrupt_source = source  # first interrupt only: later calls return above
        self._cancel()
        return self

    async def wait_for_playout(self) -> None:
        """Waits for the entire assistant turn to complete playback.

        This method waits until the assistant has fully finished speaking,
        including any finalization steps beyond initial response generation.
        This is appropriate to call when you want to ensure the speech output
        has entirely played out, including any tool calls and response follow-ups."""

        # raise an error to avoid developer mistakes
        from .agent import _get_activity_task_info

        if task := asyncio.current_task():
            info = _get_activity_task_info(task)
            if (
                info
                and info.function_call
                and info.speech_handle == self
                and not info.function_call.extra.get("__livekit_agents_tool_non_blocking", False)
            ):
                raise RuntimeError(
                    f"cannot call `SpeechHandle.wait_for_playout()` from inside the function tool `{info.function_call.name}` that owns this SpeechHandle. "
                    "This creates a circular wait: the speech handle is waiting for the function tool to complete, "
                    "while the function tool is simultaneously waiting for the speech handle.\n"
                    "To wait for the assistant’s spoken response prior to running this tool, use `RunContext.wait_for_playout()` instead."
                )

        await asyncio.shield(self._done_fut)

    def __await__(self) -> Generator[None, None, SpeechHandle]:
        async def _await_impl() -> SpeechHandle:
            await self.wait_for_playout()
            return self

        return _await_impl().__await__()

    def add_done_callback(self, callback: Callable[[SpeechHandle], None]) -> None:
        if self.done():
            asyncio.get_running_loop().call_soon(callback, self)
            return

        self._done_callbacks.add(callback)

    def remove_done_callback(self, callback: Callable[[SpeechHandle], None]) -> None:
        self._done_callbacks.discard(callback)

    async def wait_if_not_interrupted(self, aw: list[asyncio.futures.Future[Any]]) -> None:
        # wrap each future in shield so we don't cancel them when we cancel the gather future
        gather_fut = asyncio.gather(*[asyncio.shield(fut) for fut in aw], return_exceptions=True)
        fs: set[asyncio.Future[Any]] = {gather_fut, self._interrupt_fut}
        _, pending = await asyncio.wait(fs, return_when=asyncio.FIRST_COMPLETED)
        if gather_fut in pending:
            with contextlib.suppress(asyncio.CancelledError):
                gather_fut.cancel()
                await gather_fut

    def _cancel(self) -> SpeechHandle:
        if self.done():
            return self

        if not self._interrupt_fut.done():
            self._interrupt_fut.set_result(None)

            def _on_timeout() -> None:
                logger.error(
                    "speech not done in time after interruption, cancelling the speech arbitrarily.",
                    extra={"speech_id": self._id, "timeout": INTERRUPTION_TIMEOUT},
                )
                for task in self._tasks:
                    task.cancel()
                self._mark_done()

            self._interrupt_timeout_handle = asyncio.get_event_loop().call_later(
                INTERRUPTION_TIMEOUT, _on_timeout
            )

        return self

    def _add_item_added_callback(self, callback: Callable[[llm.ChatItem], Any]) -> None:
        self._item_added_callbacks.add(callback)

    def _remove_item_added_callback(self, callback: Callable[[llm.ChatItem], Any]) -> None:
        self._item_added_callbacks.discard(callback)

    def _item_added(self, items: Sequence[llm.ChatItem]) -> None:
        for item in items:
            for cb in list(self._item_added_callbacks):
                try:
                    cb(item)
                except Exception as e:
                    logger.warning(f"error in item_added_callback: {cb}", exc_info=e)

            self._chat_items.append(item)

    def _authorize_generation(self) -> None:
        fut = asyncio.Future[None]()
        self._generations.append(fut)
        if self._authorized_at is None:
            self._authorized_at = time.perf_counter()
        self._authorize_event.set()

    def _clear_authorization(self) -> None:
        self._authorize_event.clear()

    async def _wait_for_authorization(self) -> None:
        await self._authorize_event.wait()

    async def _wait_for_generation(self, step_idx: int = -1) -> None:
        if not self._generations:
            raise RuntimeError("cannot use wait_for_generation: no active generation is running.")

        await asyncio.shield(self._generations[step_idx])

    async def _wait_for_scheduled(self) -> None:
        await asyncio.shield(self._scheduled_fut)

    def _mark_generation_done(self) -> None:
        if not self._generations:
            raise RuntimeError("cannot use mark_generation_done: no active generation is running.")

        with contextlib.suppress(asyncio.InvalidStateError):
            self._generations[-1].set_result(None)

    def _mark_done(self, error: BaseException | None = None) -> None:
        # the error is kept out of _done_fut so awaiting the handle never raises
        # (most handles are never awaited); it is exposed via exception() instead
        if not self._done_fut.done():
            if error is not None:
                self._error = error
            self._done_fut.set_result(None)
        # a pipeline LLM failure is stored on the handle before the tasks finish
        self._end_agent_turn(error if error is not None else self._error)

        if self._generations:
            self._mark_generation_done()

        if self._interrupt_timeout_handle is not None:
            self._interrupt_timeout_handle.cancel()
            self._interrupt_timeout_handle = None

    def _take_agent_turn(self) -> tuple[trace.Span, float | None, str | None] | None:
        """Detach this speech's open ``agent_turn`` so a successor can continue it.

        Used when a preemptive generation is discarded for another speech answering the same
        user turn: the wasted generation stays visible under the one turn instead of becoming
        a turn of its own. After this the speech ends without touching the span."""
        span = self._agent_turn_span
        if span is None:
            return None
        carry = (span, self._agent_turn_started_at, self._agent_turn_agent_name)
        self._agent_turn_span = None
        self._agent_turn_context = None
        self._agent_turn_started_at = None
        self._agent_turn_agent_name = None
        return carry

    def _continue_agent_turn(
        self, carry: tuple[trace.Span, float | None, str | None], *, discarded: SpeechHandle
    ) -> None:
        """Adopt the ``agent_turn`` taken from ``discarded`` (see ``_take_agent_turn``)."""
        # adopted even when sampled out: the duration metric still needs the start time
        span, started_at, agent_name = carry
        span.add_event(
            "preemptive_generation_discarded", {trace_types.ATTR_SPEECH_ID: discarded.id}
        )
        span.set_attribute(trace_types.ATTR_SPEECH_ID, self.id)
        self._agent_turn_span = span
        self._agent_turn_context = trace.set_span_in_context(span)
        self._agent_turn_started_at = started_at
        self._agent_turn_agent_name = agent_name

    def _end_agent_turn(self, error: BaseException | None) -> None:
        """Close the speech's ``agent_turn`` span: the speech is done, whatever step it was on."""
        span, self._agent_turn_span = self._agent_turn_span, None
        if span is None:
            return
        from ..telemetry import otel_metrics, utils as trace_utils

        # the duration metric does not depend on the span being sampled in
        if self._agent_turn_started_at is not None and self._agent_turn_agent_name is not None:
            otel_metrics.record_invoke_agent_duration(
                time.perf_counter() - self._agent_turn_started_at,
                agent_name=self._agent_turn_agent_name,
            )
        if not span.is_recording():
            return
        if isinstance(error, Exception):
            trace_utils.record_exception(span, error)
        span.end()

    def _mark_scheduled(self) -> None:
        if self._scheduled_at is None:
            self._scheduled_at = time.perf_counter()
        with contextlib.suppress(asyncio.InvalidStateError):
            self._scheduled_fut.set_result(None)

    def _queue_wait(self) -> float | None:
        """Seconds between scheduling and the first generation authorization, once known."""
        if self._scheduled_at is None or self._authorized_at is None:
            return None
        return max(self._authorized_at - self._scheduled_at, 0.0)

class SpeechSteeringOptions(TypedDict, total=False):
    """Steers verbal delivery and non-verbal sounds in generated speech.

    Every key is a sparse override on the default (full sound vocabulary,
    light fillers): the expressive instructions already tell the LLM to match
    its delivery to the register of the moment, so most agents need no
    steering at all — set a key only to take an option away regardless of
    context.
    """

    disfluencies: bool
    """Filler words such as "um" / "uh". On by default
    (``DEFAULT_SPEECH_STEERING_OPTIONS``); set ``False`` to opt out."""
    nonverbal_sounds: bool | NonverbalOptions
    """Which non-verbal sounds the TTS may make. ``True`` (and omitting the
    key) keeps the provider's full vocabulary, ``False`` disables every sound,
    and a ``NonverbalOptions`` dict toggles per category (omitted categories
    stay enabled)."""
    pace: Literal["slow", "normal", "fast"]

class ToolCallEnded(BaseModel):
    """A tool call's single terminal entry."""

    type: Literal["tool_call_ended"] = "tool_call_ended"
    id: str
    """Entry id: ``call_id`` inline, ``{call_id}_final`` when deferred."""
    call_id: str
    message: str | None = None
    """Result or error text; None when there is nothing to voice."""
    status: Literal["done", "error", "cancelled"]

class ToolCallStarted(BaseModel):
    """A function tool call was dispatched."""

    type: Literal["tool_call_started"] = "tool_call_started"
    function_call: FunctionCall

class ToolCallUpdated(BaseModel):
    """A progress update emitted via ``ctx.update()`` while a tool call runs."""

    type: Literal["tool_call_updated"] = "tool_call_updated"
    id: str
    """Entry id: ``call_id`` inline, ``{call_id}_update_N`` when deferred."""
    call_id: str
    message: str

class ToolExecutionUpdatedEvent(BaseModel):
    """One flat tool-lifecycle update. Discriminate on ``update.type``: ``tool_call_started``
    → ``tool_call_updated`` → ``tool_call_ended`` → ``tool_reply_updated``."""

    type: Literal["tool_execution_updated"] = "tool_execution_updated"
    update: Annotated[
        ToolCallStarted | ToolCallUpdated | ToolCallEnded | ToolReplyUpdated,
        Field(discriminator="type"),
    ]
    created_at: float = Field(default_factory=time.time)

class ToolReplyUpdated(BaseModel):
    """Lifecycle of the deferred reply that voices buffered tool updates: ``scheduled``
    when queued, then ``completed`` / ``interrupted`` / ``skipped``. One reply may cover
    several calls; an inline first update never gets one."""

    type: Literal["tool_reply_updated"] = "tool_reply_updated"
    update_ids: list[str]
    """``ToolCallUpdated.id`` values this reply covers."""
    status: Literal["scheduled", "completed", "interrupted", "skipped"]
    speech_id: str
    """Id of the reply speech; ``speech_created`` carries its handle."""

class TranscriptSynchronizer:
    """
    Synchronizes text with audio playback timing.

    This class is responsible for synchronizing text with audio playback timing.
    It starts sending transcription when AudioOutput.on_playback_started is called.
    """

    def __init__(
        self,
        *,
        next_in_chain_audio: io.AudioOutput,
        next_in_chain_text: io.TextOutput | None,
        speed: float = 1.0,
        hyphenate_word: Callable[[str], list[str]] = tokenize.basic.hyphenate_word,
        word_tokenizer: NotGivenOr[tokenize.WordTokenizer] = NOT_GIVEN,
    ) -> None:
        super().__init__()

        self._text_output = _SyncedTextOutput(self, next_in_chain=next_in_chain_text)
        self._audio_output = _SyncedAudioOutput(self, next_in_chain=next_in_chain_audio)
        self._text_attached, self._audio_attached = True, True
        self._opts = _TextSyncOptions(
            speed=speed,
            hyphenate_word=hyphenate_word,
            word_tokenizer=(
                word_tokenizer
                or tokenize.basic.WordTokenizer(
                    retain_format=True, ignore_punctuation=False, split_character=True
                )
            ),
            speaking_rate_detector=SpeakingRateDetector(),
        )
        self._enabled = True
        self._closed = False

        # track pause state at the synchronizer level to apply to new impls after rotation
        self._paused = False

        # warn once per enabled cycle when only one of audio/text is detached; reset when
        # the synchronizer transitions back to enabled
        self._warned_asymmetric_detach = False

        # initial segment/first segment, recreated for each new segment
        self._impl = _SegmentSynchronizerImpl(options=self._opts, next_in_chain=next_in_chain_text)
        self._rotate_segment_atask: asyncio.Task[None] | None = None

    @property
    def audio_output(self) -> _SyncedAudioOutput:
        return self._audio_output

    @property
    def text_output(self) -> _SyncedTextOutput:
        return self._text_output

    @property
    def enabled(self) -> bool:
        return self._enabled

    async def aclose(self) -> None:
        self._closed = True
        await self.barrier()
        await self._impl.aclose()

    def set_enabled(self, enabled: bool) -> None:
        if self._enabled == enabled:
            return

        self._enabled = enabled
        if enabled:
            self._warned_asymmetric_detach = False
        if not self._rotate_segment_atask or self._rotate_segment_atask.done():
            # avoid calling rotate_segment twice when closing the session during agent speaking
            # first time when speech interrupted, second time here when output detached
            self.rotate_segment()

    def _on_attachment_changed(
        self,
        *,
        audio_attached: NotGivenOr[bool] = NOT_GIVEN,
        text_attached: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        if is_given(audio_attached):
            self._audio_attached = audio_attached

        if is_given(text_attached):
            self._text_attached = text_attached

        self.set_enabled(self._audio_attached and self._text_attached)

    async def _rotate_segment_task(self, old_task: asyncio.Task[None] | None) -> None:
        if old_task:
            with contextlib.suppress(Exception):
                await old_task

        old_impl = self._impl
        try:
            await old_impl.aclose()
        except Exception:
            logger.exception(
                "failed to close segment synchronizer impl during rotation",
                extra={"impl_id": old_impl.id},
            )

        # always create a new impl even if aclose() failed, to avoid leaving
        # self._impl pointing to a closed impl which causes the agent to get stuck
        self._impl = _SegmentSynchronizerImpl(
            options=self._opts, next_in_chain=self._text_output.next_in_chain
        )

        # apply the current pause state to the new impl
        if self._paused:
            self._impl.pause()

    def rotate_segment(self) -> None:
        if self._closed:
            return

        if self._rotate_segment_atask and not self._rotate_segment_atask.done():
            logger.warning(
                "rotate_segment called while previous segment is still being rotated",
                extra={"impl_id": self._impl.id},
            )

        self._rotate_segment_atask = asyncio.create_task(
            self._rotate_segment_task(self._rotate_segment_atask)
        )

    async def barrier(self) -> None:
        if self._rotate_segment_atask is None:
            return

        # using a while loop in case rotate_segment is called twice (this should not happen, but
        # just in case, we do log a warning if it does)
        while not self._rotate_segment_atask.done():
            await self._rotate_segment_atask

class UserInputTranscribedEvent(BaseModel):
    type: Literal["user_input_transcribed"] = "user_input_transcribed"
    transcript: str
    is_final: bool
    item_id: str | None = None
    """Provider-specific ID for the transcribed input item, when available."""
    speaker_id: str | None = None
    language: LanguageCode | None = None
    created_at: float = Field(default_factory=time.time)

class UserStateChangedEvent(BaseModel):
    type: Literal["user_state_changed"] = "user_state_changed"
    old_state: UserState
    new_state: UserState
    created_at: float = Field(default_factory=time.time)

class UserTranscriptionTimeoutEvent(BaseModel):
    type: Literal["user_transcription_timeout"] = "user_transcription_timeout"
    speech_duration: float
    """Total VAD-detected speech (s) in the turn that produced no transcript."""
    vad_speech_started_at: float
    """When VAD first detected speech for this (untranscribed) turn."""
    created_at: float = Field(default_factory=time.time)

class UserTurnExceededEvent(BaseModel):
    type: Literal["user_turn_exceeded"] = "user_turn_exceeded"
    transcript: str
    """Transcript from the current (uncommitted) user turn only.
    Previous turns in the accumulation window are already in the chat context."""
    accumulated_transcript: str
    """Full transcript since the start of user speaking."""
    accumulated_word_count: int
    """Total word count since the start of user speaking."""
    duration: float
    """Duration of the user turn in seconds."""
    created_at: float = Field(default_factory=time.time)

class VoiceActivityVideoSampler:
    def __init__(self, *, speaking_fps: float = 1.0, silent_fps: float = 0.3):
        self.speaking_fps = speaking_fps
        self.silent_fps = silent_fps
        self._last_sampled_time: float | None = None

    def __call__(self, frame: rtc.VideoFrame, session: AgentSession) -> bool:
        now = time.time()
        is_speaking = session.user_state == "speaking"
        target_fps = self.speaking_fps if is_speaking else self.silent_fps
        if target_fps == 0:
            return False
        min_frame_interval = 1.0 / target_fps

        if self._last_sampled_time is None:
            self._last_sampled_time = now
            return True

        if (now - self._last_sampled_time) >= min_frame_interval:
            self._last_sampled_time = now
            return True

        return False

class _ParticipantAudioOutput(io.AudioOutput):
    def __init__(
        self,
        room: rtc.Room,
        *,
        sample_rate: int,
        num_channels: int,
        track_publish_options: rtc.TrackPublishOptions,
        track_name: str = "roomio_audio",
    ) -> None:
        super().__init__(
            label="RoomIO",
            next_in_chain=None,
            sample_rate=sample_rate,
            capabilities=io.AudioOutputCapabilities(pause=True),
        )
        self._room = room
        self._track_name = track_name
        self._lock = asyncio.Lock()
        self._audio_source = rtc.AudioSource(sample_rate, num_channels, queue_size_ms=200)
        self._publish_options = track_publish_options
        self._publication: rtc.LocalTrackPublication | None = None
        self._subscribed_fut = asyncio.Future[None]()

        self._audio_buf = utils.aio.Chan[rtc.AudioFrame]()
        self._audio_bstream = utils.audio.AudioByteStream(
            sample_rate, num_channels, samples_per_channel=sample_rate // 20
        )
        # flushes the byte stream's tail before the source runs dry
        self._bstream_flush_timer: asyncio.TimerHandle | None = None

        self._flush_task: asyncio.Task[None] | None = None
        self._interrupted_event = asyncio.Event()
        self._forwarding_task: asyncio.Task[None] | None = None

        self._pushed_duration: float = 0.0
        self._source_pushed_duration: float = 0.0
        self._source_discarded_duration: float = 0.0
        self._interruption_generation = 0

        # the playhead sits at `_source_pushed_duration - queued_duration`; `_dry_at` is when
        # the source runs out if nothing more is pushed, which dates a run that has ended
        self._run_offset: float = 0.0
        self._dry_at: float | None = None

        self._playback_enabled = asyncio.Event()
        self._playback_enabled.set()
        # set only when _forward_audio is neither holding nor submitting a frame
        self._forwarding_idle = asyncio.Event()
        self._forwarding_idle.set()
        # playback_started fires once per segment; a mid-segment pause/resume does not re-arm this
        self._first_frame_event = asyncio.Event()

    async def _publish_track(self) -> None:
        async with self._lock:
            track = rtc.LocalAudioTrack.create_audio_track(self._track_name, self._audio_source)
            self._publication = await self._room.local_participant.publish_track(
                track, self._publish_options
            )
            await self._publication.wait_for_subscription()
            if not self._subscribed_fut.done():
                self._subscribed_fut.set_result(None)

    @property
    def subscribed(self) -> asyncio.Future[None]:
        return self._subscribed_fut

    async def start(self, *, trace_context: otel_context.Context | None = None) -> None:
        self._forwarding_task = asyncio.create_task(self._forward_audio())
        # detached: publishing spawns the track's tasks, which must not inherit this span
        with tracer.detached_span("publish_audio_output", context=trace_context) as span:
            await self._publish_track()
            if self._publication is not None:
                span.set_attribute(trace_types.ATTR_TRACK_SID, self._publication.sid)

    async def aclose(self) -> None:
        if self._bstream_flush_timer:
            self._bstream_flush_timer.cancel()
        if self._flush_task:
            await utils.aio.cancel_and_wait(self._flush_task)
        if self._forwarding_task:
            await utils.aio.cancel_and_wait(self._forwarding_task)

        await self._audio_source.aclose()

    async def capture_frame(self, frame: rtc.AudioFrame) -> None:
        await self._subscribed_fut

        await super().capture_frame(frame)

        if self._flush_task and not self._flush_task.done():
            logger.error("capture_frame called while flush is in progress")
            await self._flush_task

        for f in self._audio_bstream.push(frame.data):
            self._audio_buf.send_nowait(f)
            self._pushed_duration += f.duration
        self._arm_bstream_flush()

    def flush(self) -> None:
        super().flush()
        self._flush_bstream(force=True)

        if not self._pushed_duration:
            return

        if self._flush_task and not self._flush_task.done():
            # shouldn't happen if only one active speech handle at a time
            logger.error("flush called while playback is in progress")
            self._flush_task.cancel()

        self._flush_task = asyncio.create_task(self._wait_for_playout())

    def clear_buffer(self) -> None:
        if self._bstream_flush_timer:
            self._bstream_flush_timer.cancel()
            self._bstream_flush_timer = None
        self._audio_bstream.clear()

        if not self._pushed_duration:
            return

        # a detaching sink never reaches the playout wait, so the playhead is reported here
        self._report_run(
            offset=self._source_pushed_duration - self._audio_source.queued_duration,
            ended_at=time.time(),
        )
        self._interrupted_event.set()

    def pause(self) -> None:
        super().pause()
        self._playback_enabled.clear()
        if self._bstream_flush_timer:
            self._bstream_flush_timer.cancel()
            self._bstream_flush_timer = None
        # self._audio_source.clear_queue()

    def resume(self) -> None:
        super().resume()
        self._playback_enabled.set()
        self._arm_bstream_flush()

    def _arm_bstream_flush(self, *, margin: float = 0.02) -> None:
        # audio short of a frame waits for more, and the timer flushes it `margin` seconds before
        # the output runs dry; a paused output needs no frames, so no timer runs while paused
        if self._bstream_flush_timer:
            self._bstream_flush_timer.cancel()
            self._bstream_flush_timer = None
        if not self._playback_enabled.is_set():
            return
        queued = (
            self._audio_source.queued_duration
            + self._pushed_duration
            - self._source_pushed_duration
        )
        self._bstream_flush_timer = asyncio.get_running_loop().call_later(
            max(queued - margin, 0.0), self._flush_bstream
        )

    def _flush_bstream(self, *, force: bool = False) -> None:
        if self._bstream_flush_timer:
            self._bstream_flush_timer.cancel()
            self._bstream_flush_timer = None
        # nothing plays before the first frame, so one under 20 ms waits for more instead
        if not force and not self._pushed_duration and self._audio_bstream.buffered_duration < 0.02:
            return
        for f in self._audio_bstream.flush():
            self._audio_buf.send_nowait(f)
            self._pushed_duration += f.duration

    def _report_run(
        self, *, offset: float, ended_at: float, resumes_at: float | None = None
    ) -> None:
        """Report the open run, and begin the next past any audio that never played.

        The run covers ``[_run_offset, offset)`` and stopped at ``ended_at``. The next one
        starts at ``offset``, or at ``resumes_at`` when the audio between the two was
        discarded rather than played.
        """
        if self._dry_at is not None:
            # a run cannot outlast the audio the source held, however late the caller noticed
            ended_at = min(ended_at, self._dry_at)

        duration = offset - self._run_offset
        if duration > 0:
            self.on_playback_progressed(
                started_at=ended_at - duration, offset=self._run_offset, duration=duration
            )
        self._run_offset = offset if resumes_at is None else resumes_at
        self._dry_at = None

    async def _wait_for_playout(self) -> None:
        wait_for_interruption = asyncio.create_task(self._interrupted_event.wait())

        async def _wait_buffered_audio() -> None:
            while True:
                await self._forwarding_idle.wait()
                if self._audio_buf.empty():
                    break

                await self._playback_enabled.wait()
                await asyncio.sleep(0)

            await self._audio_source.wait_for_playout()

        wait_for_playout = asyncio.create_task(_wait_buffered_audio())
        await asyncio.wait(
            [wait_for_playout, wait_for_interruption],
            return_when=asyncio.FIRST_COMPLETED,
        )

        interrupted = self._interrupted_event.is_set()
        pushed_duration = max(
            self._source_pushed_duration - self._source_discarded_duration,
            0,
        )

        if interrupted:
            self._interruption_generation += 1
            queued_duration = self._audio_source.queued_duration
            while not self._audio_buf.empty():
                self._audio_buf.recv_nowait()

            pushed_duration = max(pushed_duration - queued_duration, 0)
            # the playhead stops where the cleared queue begins
            self._report_run(
                offset=self._source_pushed_duration - queued_duration, ended_at=time.time()
            )
            self._audio_source.clear_queue()
            wait_for_playout.cancel()
        else:
            wait_for_interruption.cancel()
            # the source drained, so everything pushed has played
            self._report_run(offset=self._source_pushed_duration, ended_at=time.time())

        self._run_offset = 0.0

        self._pushed_duration = 0
        self._source_pushed_duration = 0
        self._source_discarded_duration = 0
        self._interrupted_event.clear()
        self._first_frame_event.clear()
        self.on_playback_finished(playback_position=pushed_duration, interrupted=interrupted)

    async def _forward_audio(self) -> None:
        async for frame in self._audio_buf:
            interruption_generation = self._interruption_generation
            self._forwarding_idle.clear()
            try:
                if not self._playback_enabled.is_set():
                    queued = self._audio_source.queued_duration
                    self._source_discarded_duration += queued
                    # the dropped queue never plays, so the next run resumes past it
                    self._report_run(
                        offset=self._source_pushed_duration - queued,
                        ended_at=time.time(),
                        resumes_at=self._source_pushed_duration,
                    )
                    self._audio_source.clear_queue()
                    await self._playback_enabled.wait()
                    # drop a paused frame when its original segment was interrupted.
                    if interruption_generation != self._interruption_generation:
                        continue
                    # TODO(long): ignore frames from previous syllable

                if self._interrupted_event.is_set() or self._pushed_duration == 0:
                    if self._interrupted_event.is_set() and self._flush_task:
                        await self._flush_task

                    # ignore frames if interrupted
                    continue

                if not self._first_frame_event.is_set():
                    self._first_frame_event.set()
                    self.on_playback_started(created_at=time.time())
                if self._dry_at is not None and time.time() >= self._dry_at:
                    # the source ran out before this frame arrived
                    self._report_run(offset=self._source_pushed_duration, ended_at=self._dry_at)

                self._source_pushed_duration += frame.duration
                await self._audio_source.capture_frame(frame)
                self._dry_at = time.time() + self._audio_source.queued_duration
            finally:
                self._forwarding_idle.set()

class _ParticipantStreamTranscriptionOutput:
    def __init__(
        self,
        room: rtc.Room,
        *,
        is_delta_stream: bool = True,
        participant: rtc.Participant | str | None = None,
        attributes: dict[str, str] | None = None,
        json_format: bool = False,
    ):
        self._room, self._is_delta_stream = room, is_delta_stream
        self._track_id: str | None = None
        self._participant_identity: str | None = None
        self._additional_attributes = attributes or {}
        self._writer: rtc.TextStreamWriter | None = None
        self._json_format = json_format

        self._room.on("track_published", self._on_track_published)
        self._room.on("local_track_published", self._on_local_track_published)
        self._flush_atask: asyncio.Task[None] | None = None
        self._closed = False

        self._reset_state()
        self.set_participant(participant)

    def set_participant(
        self,
        participant: rtc.Participant | str | None,
    ) -> None:
        self._participant_identity = (
            participant.identity if isinstance(participant, rtc.Participant) else participant
        )
        if self._participant_identity is None:
            return

        try:
            self._track_id = find_micro_track_id(self._room, self._participant_identity)
        except ValueError:
            # track id is optional for TextStream when audio is not published
            self._track_id = None

        self.flush()
        self._reset_state()

    def _reset_state(self) -> None:
        self._current_id = utils.shortuuid("SG_")
        self._capturing = False
        self._latest_text = ""
        # per-segment markup stripping: delta streams strip incrementally (buffering a tag
        # split across chunks); non-delta streams re-strip the full text each time and keep
        # the latest tags here for the expression attribute (see TranscriptMarkupStripper)
        self._stripper = TranscriptMarkupStripper()
        self._segment_tags: list[ExpressiveTag] = []

    def _encode(self, clean_text: str, timing_src: str | None = None) -> str:
        """Wrap visible text for the wire (JSON TimedString when json_format, else raw)."""
        if not self._json_format:
            return clean_text

        ts_pb = agent_pb.TimedString(text=clean_text)
        if isinstance(timing_src, TimedString):
            if utils.is_given(timing_src.start_time):
                ts_pb.start_time = timing_src.start_time
            if utils.is_given(timing_src.end_time):
                ts_pb.end_time = timing_src.end_time
            if utils.is_given(timing_src.confidence):
                ts_pb.confidence = timing_src.confidence
            if utils.is_given(timing_src.start_time_offset):
                ts_pb.start_time_offset = timing_src.start_time_offset
        return json.dumps(MessageToDict(ts_pb, preserving_proto_field_name=True)) + "\n"

    async def _create_text_writer(
        self,
        attributes: dict[str, str] | None = None,
        extra: dict[str, str] | None = None,
    ) -> rtc.TextStreamWriter:
        assert self._participant_identity is not None, "participant_identity is not set"

        if not attributes:
            attributes = {
                ATTRIBUTE_TRANSCRIPTION_FINAL: "false",
            }
            if self._track_id:
                attributes[ATTRIBUTE_TRANSCRIPTION_TRACK_ID] = self._track_id
        attributes[ATTRIBUTE_TRANSCRIPTION_SEGMENT_ID] = self._current_id
        # overlaid rather than replacing, so the caller can add a key without
        # dropping the transcription attributes the protocol requires
        if extra:
            attributes.update(extra)

        for key, val in self._additional_attributes.items():
            if key not in attributes:
                attributes[key] = val

        return await self._room.local_participant.stream_text(
            topic=TOPIC_TRANSCRIPTION,
            sender_identity=self._participant_identity,
            attributes=attributes,
        )

    @utils.log_exceptions(logger=logger)
    async def capture_text(self, text: str) -> None:
        if self._participant_identity is None:
            return

        if self._flush_atask and not self._flush_atask.done():
            await self._flush_atask

        if not self._capturing:
            self._reset_state()
            self._capturing = True

        # the raw text (expressive markup intact) arrives here; publish only the visible
        # text. Skip a chunk that strips to nothing (a partial tag still buffering, or a
        # markup-only token) so the transcript cadence isn't disturbed.
        if self._is_delta_stream:
            clean_text = self._stripper.push(text)
        else:
            clean_text, self._segment_tags = split_all_markup(text)
        if not clean_text:
            return

        payload = self._encode(clean_text, text)
        self._latest_text = payload

        try:
            if self._room.isconnected():
                if self._is_delta_stream:  # reuse the existing writer
                    if self._writer is None:
                        # the leading expression is stripped before any visible text, so it is
                        # already known here. Put it on the opening header as well as the closing
                        # one, or a frontend can't colour the turn until the agent stops talking.
                        self._writer = await self._create_text_writer(
                            extra=expression_attribute(self._stripper.tags)
                        )

                    await self._writer.write(payload)
                else:  # always create a new writer
                    tmp_writer = await self._create_text_writer()
                    await tmp_writer.write(payload)
                    await tmp_writer.aclose()
        except Exception as e:
            logger.warning("failed to publish agent transcription to room: %s", e)

    async def _flush_task(
        self,
        writer: rtc.TextStreamWriter | None,
        extra_attributes: dict[str, str] | None = None,
        pending_text: str = "",
    ) -> None:
        attributes = {ATTRIBUTE_TRANSCRIPTION_FINAL: "true"}
        if self._track_id:
            attributes[ATTRIBUTE_TRANSCRIPTION_TRACK_ID] = self._track_id
        for key, val in (extra_attributes or {}).items():
            attributes.setdefault(key, val)

        try:
            if self._room.isconnected():
                if self._is_delta_stream:
                    if writer:
                        if pending_text:  # visible text left in the strip buffer
                            await writer.write(pending_text)
                        await writer.aclose(attributes=attributes)
                else:
                    tmp_writer = await self._create_text_writer(attributes=attributes)
                    await tmp_writer.write(self._latest_text)
                    await tmp_writer.aclose()
        except Exception as e:
            logger.warning("failed to publish agent transcription to room: %s", e)

    def flush(self) -> None:
        # only emit on a segment that captured text (keeps lk.transcription cadence intact).
        # The leading expression the sinks stripped rides along on the closing header as the
        # lk.expression attribute.
        if self._participant_identity is None or not self._capturing:
            return

        self._capturing = False
        curr_writer = self._writer
        self._writer = None

        if self._is_delta_stream:
            remaining = self._stripper.flush()
            tags = self._stripper.tags
        else:
            remaining = ""
            tags = self._segment_tags

        pending_text = self._encode(remaining) if remaining else ""
        self._flush_atask = asyncio.create_task(
            self._flush_task(curr_writer, expression_attribute(tags), pending_text)
        )

    async def aclose(self) -> None:
        if self._closed:
            return

        self._closed = True
        self._room.off("track_published", self._on_track_published)
        self._room.off("local_track_published", self._on_local_track_published)

        if self._flush_atask:
            await self._flush_atask

        if self._writer:
            writer = self._writer
            self._writer = None
            await writer.aclose()

    def _on_track_published(
        self, track: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant
    ) -> None:
        if (
            self._participant_identity is None
            or participant.identity != self._participant_identity
            or track.source != rtc.TrackSource.SOURCE_MICROPHONE
        ):
            return

        self._track_id = track.sid

    def _on_local_track_published(self, track: rtc.LocalTrackPublication, _: rtc.Track) -> None:
        if (
            self._participant_identity is None
            or self._participant_identity != self._room.local_participant.identity
            or track.source != rtc.TrackSource.SOURCE_MICROPHONE
        ):
            return

        self._track_id = track.sid

class _ParticipantTranscriptionOutput(io.TextOutput):
    def __init__(
        self,
        *,
        room: rtc.Room,
        is_delta_stream: bool = True,
        participant: rtc.Participant | str | None = None,
        next_in_chain: io.TextOutput | None = None,
        json_format: bool = False,
    ) -> None:
        super().__init__(label="RoomIO", next_in_chain=next_in_chain)

        self.__outputs: list[
            _ParticipantLegacyTranscriptionOutput | _ParticipantStreamTranscriptionOutput
        ] = [
            _ParticipantLegacyTranscriptionOutput(
                room=room,
                is_delta_stream=is_delta_stream,
                participant=participant,
            ),
            _ParticipantStreamTranscriptionOutput(
                room=room,
                is_delta_stream=is_delta_stream,
                participant=participant,
                json_format=json_format,
            ),
        ]
        self.__closed = False

    def set_participant(self, participant: rtc.Participant | str | None) -> None:
        for source in self.__outputs:
            source.set_participant(participant)

    async def capture_text(self, text: str) -> None:
        await asyncio.gather(*[sink.capture_text(text) for sink in self.__outputs])

        if self.next_in_chain:
            await self.next_in_chain.capture_text(text)

    def flush(self) -> None:
        for source in self.__outputs:
            source.flush()

        if self.next_in_chain:
            self.next_in_chain.flush()

    async def aclose(self) -> None:
        if self.__closed:
            return

        self.__closed = True
        await asyncio.gather(*[source.aclose() for source in self.__outputs])