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 | LLMModels | str | None] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | TTSModels | str | None] = 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
        self._llm = llm
        self._tts = tts
        self._vad = vad

        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
        )

    # -- 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

            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"
                    )

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

            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)

        @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

            if not activity.tts.capabilities.streaming:
                wrapped_tts = 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,
                    ),
                )

            # 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)

        @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 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 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",
        },
        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 | LLMModels | str] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | TTSModels | str] = NOT_GIVEN,
        turn_handling: NotGivenOr[TurnHandlingOptions] = NOT_GIVEN,
        keyterms_options: NotGivenOr[KeytermsOptions] = 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,
        # Misc settings
        userdata: NotGivenOr[Userdata_T] = NOT_GIVEN,
        video_sampler: NotGivenOr[_VideoSampler | None] = NOT_GIVEN,
        aec_warmup_duration: float | None = 3.0,
        ivr_detection: bool = False,
        user_away_timeout: float | None = 15.0,
        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,
    ) -> 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.
            keyterms_options (KeytermsOptions, optional): Keyterm biasing for the STT. Holds
                static ``keyterms`` plus ``keyterm_detection`` (LLM extraction). Applies to STTs
                that accept a term list; on others it warns and is ignored.
            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.
            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.
            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.
                Set to ``None`` to disable. Default ``3.0`` s.
            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"))
        user_turn_limit = _resolve_user_turn_limit(turn_handling.get("user_turn_limit"))

        # 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,
            ),
            keyterms_options=_resolve_keyterms_options(keyterms_options or None),
            endpointing_overrides=endpointing_overrides,
            max_tool_steps=max_tool_steps,
            user_away_timeout=user_away_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=aec_warmup_duration,
            session_close_transcript_timeout=session_close_transcript_timeout,
        )
        # expressive mode is not publicly exposed; the pipeline stays disabled
        self._expressive: bool | ExpressiveOptions = False
        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
        self._llm = llm or None
        self._tts = tts or None

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

        self._turn_detection = raw_turn_detection
        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, reset after agent speaking
        self._llm_error_counts = 0
        self._tts_error_counts = 0

        # aec warmup: disable interruptions while AEC warms up
        self._aec_warmup_remaining = 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
        # TODO(theomonnom): need a better way to expose early assistant metrics
        self._early_assistant_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
        self._session_ctx_token: Token[otel_context.Context] | None = None

        self._recorded_events: list[AgentEvent] = []
        self._recording_options: RecordingOptions = _RECORDING_ALL_OFF.copy()
        self._started_at: float | None = None
        self._usage_collector = ModelUsageCollector()

        # 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,
        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,
        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,
        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
            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._recording_options = _resolve_recording_options(record)  # type: ignore[arg-type]
            if self._text_only:
                self._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 any(self._recording_options.values()):
                        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._recording_options = _resolve_recording_options(False)

                job_ctx.init_recording(self._recording_options)

            # 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._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)
                await self._room_io.start()

                if is_primary:
                    # 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._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_attribute(trace_types.ATTR_ROOM_NAME, job_ctx.room.name)
                current_span.set_attribute(trace_types.ATTR_JOB_ID, job_ctx.job.id)
                current_span.set_attribute(trace_types.ATTR_AGENT_NAME, job_ctx.job.agent_name)
                if self._room_io:
                    # automatically connect to the room when room io is used
                    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
            tasks.append(
                asyncio.create_task(self._update_activity(self._agent, wait_on_enter=False))
            )

            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()

            # 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)

    @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:
        if self._root_span_context:
            # make `activity.drain` and `on_exit` under the root span
            otel_context.attach(self._root_span_context)

        async with self._lock:
            if not self._started:
                return

            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,
                )

            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._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 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,
        # 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.
            min_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
            max_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
        """
        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

        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={"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 = trace.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.
        """  # 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 = trace.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.

        Returns:
            An asyncio.Future that completes when the interruption is fully processed
            and chat context has been updated.
        """
        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) -> AsyncIterator[AgentActivity]:
        """Wait for idle, then block other ``wait_for_idle`` callers until exit."""
        from .agent_activity import _IdleHoldContextVar

        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()

    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)

            reuse_resources: _ReusableResources | None = None
            try:
                previous_activity_v = self._activity
                if (activity := self._activity) is not None:
                    if previous_activity == "close":
                        reuse_resources = await activity.drain(new_activity=self._next_activity)
                        await activity.aclose()
                    elif previous_activity == "pause":
                        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)
                elif new_activity == "resume":
                    await self._activity.resume(reuse_resources=reuse_resources)
            except BaseException:
                if reuse_resources is not None:
                    await reuse_resources.cleanup()
                raise

        # 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)

    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 == "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 (
            (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 _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.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.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 self.user_state == "away" and ev.is_final:
            # reset user state from away to listening in case VAD has a miss detection
            self._update_user_state("listening")

        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, "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 _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 get_job_context

        job_ctx = get_job_context(required=False)
        if job_ctx is None or (sim_ctx := job_ctx.simulation_context()) is None:
            return False

        from ..simulation import SimulationMode

        return sim_ctx.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
            )

        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."
                )
            )

        current_task.add_done_callback(_handle_task_done)

        from .agent_activity import _AgentActivityContextVar, _SpeechHandleContextVar

        # TODO(theomonnom): add a global lock for inline tasks
        # This may currently break in the case we use parallel tool calls.

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

        old_allow_interruptions = True
        if speech_handle:
            if speech_handle.interrupted:
                raise RuntimeError(
                    f"{self.__class__.__name__} cannot be awaited inside a function tool that is already interrupted"
                )

            # lock the speech handle to prevent interruptions until the task is complete
            # there should be no await before this line to avoid race conditions
            old_allow_interruptions = speech_handle.allow_interruptions
            speech_handle.allow_interruptions = False

        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)

        # register before any await so a concurrent drain (e.g. session close)
        # won't wait for tasks blocked on this handoff
        old_activity._add_drain_blocked_tasks(blocked_tasks)

        # watch the blocked tasks so an active run won't complete mid-handoff
        # (the parent speech may predate the run, e.g. created in on_enter)
        if (run_state := session._global_run_state) and not run_state.done():
            for task in blocked_tasks:
                run_state._watch_handle(task)

        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.Task[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.
            if run_state and not run_state.done():
                if speech_handle and run_state._unwatch_handle(speech_handle):
                    suspended_handles.append(speech_handle)
                for task in blocked_tasks:
                    if run_state._unwatch_handle(task):
                        suspended_handles.append(task)
                if suspended_handles:
                    run_state._mark_done_if_needed(None)
        except Exception:
            self.__inactive_ev.set()
            raise

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

        finally:
            if speech_handle:
                with contextlib.suppress(RuntimeError):
                    speech_handle.allow_interruptions = old_allow_interruptions

            # 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,
        using_default_vad: bool,
        interruption_detection: inference.AdaptiveInterruptionDetector | None,
        turn_detection: TurnDetectionMode | None,
        stt_model: str | None = None,
        stt_provider: str | None = None,
    ) -> 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._using_default_vad = using_default_vad
        self._stt_model = stt_model
        self._stt_provider = stt_provider
        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._ignore_user_transcript_until: NotGivenOr[float] = NOT_GIVEN
        self._transcript_buffer: deque[SpeechEvent] = deque()
        self._interruption_enabled: bool = interruption_detection is not None and vad is not None
        self._agent_speaking: bool = False
        self._agent_speech_started_at: float | None = None

        _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
        self._stt_request_ids: list[str] = []
        self._closing = asyncio.Event()
        self.__stt_context: BaseModel | None = None

        self._vad_speech_started: 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
                    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

    @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:
        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]

    def _on_end_of_agent_speech(self, *, ignore_user_transcript_until: float) -> None:
        self._cancel_backchannel_boundary()

        if self._agent_speaking:
            self._endpointing.on_end_of_agent_speech(ended_at=time.time())

        if not self._adaptive_interruption_active:
            self._agent_speaking = False
            return

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

        if self._agent_speaking:
            # no interruption is detected, end the inference (idempotent)
            if not is_given(self._ignore_user_transcript_until):
                self._on_end_of_overlap_speech(ended_at=time.time())

            end_cooldown: float = (
                self._backchannel_boundary[1] if self._backchannel_boundary else 0.0
            )

            ignore_until = (
                ignore_user_transcript_until
                if not is_given(self._ignore_user_transcript_until)
                else min(ignore_user_transcript_until, self._ignore_user_transcript_until)
            )
            logger.trace(
                "flushing held transcripts",
                extra={
                    "ignore_until": ignore_until,
                    "end_cooldown": end_cooldown,
                },
            )
            self._ignore_user_transcript_until = ignore_until - end_cooldown

            # flush held transcripts if possible
            task = asyncio.create_task(self._flush_held_transcripts(cooldown=end_cooldown))
            task.add_done_callback(lambda _: self._tasks.discard(task))
            self._tasks.add(task)

        self._agent_speaking = False

    def _on_start_of_speech(
        self,
        started_at: float,
        speech_duration: float = 0.0,
        user_speaking_span: trace.Span | None = None,
    ) -> None:
        self._endpointing.on_start_of_speech(
            started_at=started_at, overlapping=self._agent_speaking
        )
        if not self._adaptive_interruption_active or not self._agent_speaking:
            return
        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:
        should_ignore = is_given(interruption) and not interruption and self._agent_speaking
        if self._speaking:
            self._endpointing.on_end_of_speech(
                ended_at=ended_at,
                should_ignore=should_ignore,
            )

        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,
    ) -> None:
        """End interruption inference when agent is speaking and overlap speech ends."""
        if not self._adaptive_interruption_active or not self._agent_speaking:
            return

        # 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())
        )

    @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()

    @utils.log_exceptions(logger=logger)
    async def _flush_held_transcripts(self, cooldown: float, force: bool = False) -> None:
        """Flush held transcripts.

        When ``force`` is True, all buffered events are emitted unconditionally; this
        is used during interruption-detector teardown when the ignore-window gating
        can no longer be trusted.

        Otherwise, drop transcripts whose *end time* falls before
        ``ignore_user_transcript_until - cooldown`` and re-emit the rest. Events
        without timestamps are treated as the next valid event.
        """
        if not self._transcript_buffer:
            self._reset_interruption_detection()
            return

        if force:
            events_to_emit = list(self._transcript_buffer)
            # reset before emitting to avoid recursive calls
            self._reset_interruption_detection()
            for ev in events_to_emit:
                await self._on_stt_event(ev)
            return

        if (
            not self._interruption_enabled
            or not is_given(self._ignore_user_transcript_until)
            or self._input_started_at is None
        ):
            self._reset_interruption_detection()
            return

        emit_from_index: int | None = None
        should_flush = False
        for i, ev in enumerate(self._transcript_buffer):
            # always try to emit from a sentinel event
            if not ev.alternatives:
                emit_from_index = min(emit_from_index, i) if emit_from_index is not None else i
                continue
            if ev.alternatives[0].start_time == ev.alternatives[0].end_time == 0:
                self._reset_interruption_detection()
                return

            if ev.alternatives[0].end_time > 0 and self._within_ignore_window(
                ev.alternatives[0].end_time + self._input_started_at
            ):
                # reset the index to emit from the next valid event
                emit_from_index = None
            else:
                # break since we found a valid event to emit from
                emit_from_index = min(emit_from_index, i) if emit_from_index is not None else i
                should_flush = True
                break

        events_to_emit = (
            list(self._transcript_buffer)[int(emit_from_index) :]
            if emit_from_index is not None and should_flush
            else []
        )
        _ignore_user_transcript_until = self._ignore_user_transcript_until
        _input_started_at = self._input_started_at
        # reset before emitting to avoid recursive calls
        self._reset_interruption_detection()

        for ev in events_to_emit:
            added_delay = 0.0
            if ev.alternatives and ev.alternatives[0].end_time > 0:
                added_delay = max(
                    0,
                    (
                        ev.alternatives[0].end_time
                        + _input_started_at
                        - _ignore_user_transcript_until
                    )
                    + (cooldown or 0.0),
                )
            logger.trace(
                "re-emitting held user transcript",
                extra={
                    "event": ev.type,
                    "cooldown": cooldown,
                    "added_delay": added_delay,
                },
            )
            await self._on_stt_event(ev)

    def _reset_interruption_detection(self) -> None:
        """Reset relevant states for adaptive interruption detection."""
        self._transcript_buffer.clear()
        self._ignore_user_transcript_until = NOT_GIVEN
        # keep the anchor while a newer agent-speech cycle is active, so a stale flush
        # can't clear an anchor that cycle has already set
        if not self._agent_speaking:
            self._agent_speech_started_at = None

    def _within_ignore_window(self, event_time: float) -> bool:
        """Whether a wall-clock event time falls inside the active ignore-user-transcript window."""
        if not is_given(self._ignore_user_transcript_until):
            return False
        lower = self._agent_speech_started_at or 0.0
        upper = min(time.time(), self._ignore_user_transcript_until)
        return lower < event_time < upper

    def _should_hold_stt_event(self, ev: stt.SpeechEvent) -> bool:
        """Test if the event should be held until the ignore_user_transcript_until timestamp."""
        if not self._interruption_enabled:
            return False

        if self._agent_speaking:
            return True

        # reset when the user starts speaking after the agent speech
        # this could let a transcript pass through if the user starts
        # speaking right before the agent speech ends, not ideal but
        # better than swallowing the transcript.
        if ev.type == stt.SpeechEventType.START_OF_SPEECH:
            self._ignore_user_transcript_until = NOT_GIVEN
            return False

        if not is_given(self._ignore_user_transcript_until):
            return False
        # sentinel events are always held until
        # we have something concrete to release them
        if not ev.alternatives:
            return True
        if (
            # most vendors don't set timestamps properly, in which case we just assume
            # it is a valid event after the ignore_user_transcript_until timestamp
            is_given(self._input_started_at)
            # check if the event should be held if
            # 1. the stt input stream has started
            # 2. the current event has a valid start and end time, relative to the input stream start time
            # 3. the event's wall-clock time falls inside the bounded ignore-user-transcript window
            and self._input_started_at is not None
            and not (ev.alternatives[0].start_time == ev.alternatives[0].end_time == 0)
            and ev.alternatives[0].start_time > 0
            and self._within_ignore_window(ev.alternatives[0].start_time + self._input_started_at)
        ):
            return True

        return False

    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()
        if self._commit_user_turn_atask is not None:
            await aio.cancel_and_wait(self._commit_user_turn_atask)

        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:
            await aio.cancel_and_wait(self._end_of_turn_task)

        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

    def _update_stt(self, stt: io.STTNode | None, *, pipeline: _STTPipeline | None = None) -> None:
        self._stt = stt
        if pipeline is None and stt is not None:
            pipeline = _STTPipeline(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()
            self._ignore_user_transcript_until = NOT_GIVEN
        else:
            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,
    ) -> None:
        if not is_given(detector):
            detector = self._turn_detector
        if not isinstance(detector, _StreamingTurnDetector) or self._vad is None:
            return
        if (current := getattr(self._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._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._interruption_detection = interruption_detection
        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
                )
            )
            self._transcript_buffer.clear()
            self._ignore_user_transcript_until = NOT_GIVEN
        elif 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()
            flush_task = asyncio.create_task(self._flush_held_transcripts(cooldown=0.0, force=True))
            flush_task.add_done_callback(lambda _: self._tasks.discard(flush_task))
            self._tasks.add(flush_task)

        self._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
        if self._user_turn_span is not None and self._user_turn_span.is_recording():
            self._user_turn_span.end()
        self._user_turn_span = None
        self._stt_request_ids = []

        # 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,
                                "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:
        # 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

        # handle interruption detection
        # - hold the event until the ignore_user_transcript_until expires
        # - release only relevant events
        # - allow RECOGNITION_USAGE to pass through immediately
        if ev.type != stt.SpeechEventType.RECOGNITION_USAGE and self._interruption_enabled:
            if self._should_hold_stt_event(ev):
                logger.trace(
                    "holding STT event until ignore_user_transcript_until expires",
                    extra={
                        "event": ev.type,
                        "ignore_user_transcript_until": self._ignore_user_transcript_until
                        if is_given(self._ignore_user_transcript_until)
                        else None,
                    },
                )
                self._transcript_buffer.append(ev)
                return

            if self._transcript_buffer:
                end_cooldown: float = (
                    self._backchannel_boundary[1] if self._backchannel_boundary else 0.0
                )
                await self._flush_held_transcripts(cooldown=end_cooldown)
                # no return here to allow the new event to be processed normally

        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
        )
        if ev.type == stt.SpeechEventType.FINAL_TRANSCRIPT:
            transcript = ev.alternatives[0].text
            language = ev.alternatives[0].language
            confidence = ev.alternatives[0].confidence

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

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

            self._hooks.on_final_transcript(
                ev,
                speaking=self._speaking
                if (self._vad is not None and not self._using_default_vad)
                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] = {"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 self._vad is None or self._using_default_vad or self._last_speaking_time is None:
                # vad disabled or missed a speech, use stt timestamp
                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 and not self._using_default_vad)
                or self._turn_detection_mode == "stt"
                else None,
            )
            transcript = ev.alternatives[0].text
            language = ev.alternatives[0].language
            confidence = ev.alternatives[0].confidence

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

            if not transcript:
                return

            logger.debug(
                "received user preflight transcript",
                extra={"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 self._vad is None or self._using_default_vad or self._last_speaking_time is None:
                # vad disabled or missed a speech, use stt timestamp
                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 and not self._using_default_vad)
                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 trace.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
            if self._vad is None or self._using_default_vad or self._last_speaking_time is None:
                # vad disabled or missed a speech, use stt timestamp
                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()

            with trace.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
            if not self._vad_speech_started:
                self._speech_start_time = speech_start_time
                self._vad_speech_started = True

            with trace.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:
            with trace.use_span(self._ensure_user_turn_span()):
                self._hooks.on_end_of_speech(ev)

            self._vad_speech_started = False
            self._speaking = False
            self._last_speaking_time = time.time() - 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)

    async def _on_overlap_speech_event(self, ev: inference.OverlappingSpeechEvent) -> None:
        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:
            self._hooks.on_interruption(ev)

    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
            user_turn_span = self._ensure_user_turn_span()

            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:
                    with (
                        trace.use_span(user_turn_span),
                        tracer.start_as_current_span("eou_detection") as 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,
                            )

            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(),
            )
            committed = self._hooks.on_end_of_turn(
                _EndOfTurnInfo(
                    skip_reply=skip_reply,
                    new_transcript=self._audio_transcript,
                    transcript_confidence=confidence_avg,
                    metrics=metrics,
                )
            )
            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
                    )
                user_turn_span.end()
                self._user_turn_span = None
                self._user_turn_start = None
                self._stt_request_ids = []

                # 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

            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 trace.use_span(self._ensure_user_turn_span()):
                    self._hooks.on_end_of_speech(None)
                self._speaking = False
                self._vad_speech_started = False

    @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:
                await self._on_overlap_speech_event(ev)
        except APIError:
            # avoid already emitted error from the stream
            return
        finally:
            await aio.cancel_and_wait(forward_task)
            await stream.aclose()

    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)
        self._user_turn_span = tracer.start_span("user_turn", 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

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 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. When an
    output is present, its ``call_id`` matches the paired function call's
    ``call_id``. A ``None`` output means the function call did not produce a
    value that should be sent back to the LLM, such as when a tool raises
    ``StopResponse`` or returns an invalid output.
    """

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

    def zipped(self) -> list[tuple[FunctionCall, FunctionCallOutput | None]]:
        """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:
        self._reply_required = False

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

    @property
    def has_tool_reply(self) -> bool:
        return self._reply_required

    @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.

    Can be passed as a plain dict::

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

    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 RecordingOptions(TypedDict, total=False):
    """Granular control over which recording features are active.

    All keys default to ``True`` when not specified, so ``{"logs": False}``
    means "record everything except logs."

    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
    """

    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``."""

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)

    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:
        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:
            self._pending_requests.pop(request.request_id, None)
            logger.warning(
                "remote session request timed out",
                extra={"request_id": request.request_id, "type": req_type, "timeout": timeout},
            )
            raise
        except Exception:
            self._pending_requests.pop(request.request_id, None)
            raise

        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
        while True:
            remaining = deadline - asyncio.get_event_loop().time()
            if remaining <= 0:
                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

    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_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,
    ) -> None:
        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

    @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.

        Delegates to the SpeechHandle.allow_interruptions setter,
        which will raise a RuntimeError if the handle is already interrupted.

        Raises:
            RuntimeError: If the SpeechHandle is already interrupted.
        """
        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() 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.emit(
            "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)
        # fall back to a stub when the message isn't a valid tool output (e.g. raw object)
        if tool_output.fnc_call_out is None:
            fnc_call_out = FunctionCallOutput(
                name=fnc_call.name,
                call_id=fnc_call.call_id,
                output=str(message or ""),
                is_error=False,
            )
        else:
            fnc_call_out = tool_output.fnc_call_out
        return (fnc_call, 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 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._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
        self._agent_turn_context: otel_context.Context | None = None

        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

    @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) -> SpeechHandle:
        """Interrupt the current speech generation.

        Raises:
            RuntimeError: If this speech handle does not allow interruptions.

        Returns:
            SpeechHandle: The same speech handle that was interrupted.
        """
        if not force and not self._allow_interruptions:
            raise RuntimeError("This generation handle does not allow interruptions")

        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)
        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)

        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 _mark_scheduled(self) -> None:
        with contextlib.suppress(asyncio.InvalidStateError):
            self._scheduled_fut.set_result(None)

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 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, progressive=True
        )

        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._playback_enabled = asyncio.Event()
        self._playback_enabled.set()
        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) -> None:
        self._forwarding_task = asyncio.create_task(self._forward_audio())
        await self._publish_track()

    async def aclose(self) -> None:
        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

    def flush(self) -> None:
        super().flush()

        for f in self._audio_bstream.flush():
            self._audio_buf.send_nowait(f)
            self._pushed_duration += f.duration

        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:
        self._audio_bstream.clear()

        if not self._pushed_duration:
            return
        self._interrupted_event.set()

    def pause(self) -> None:
        super().pause()
        self._playback_enabled.clear()
        # self._audio_source.clear_queue()

    def resume(self) -> None:
        super().resume()
        self._playback_enabled.set()
        self._first_frame_event.clear()

    async def _wait_for_playout(self) -> None:
        wait_for_interruption = asyncio.create_task(self._interrupted_event.wait())

        async def _wait_buffered_audio() -> None:
            while not self._audio_buf.empty():
                if not self._playback_enabled.is_set():
                    await self._playback_enabled.wait()

                await self._audio_source.wait_for_playout()
                # avoid deadlock when clear_buffer called before capture_frame
                await asyncio.sleep(0)

        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 = self._pushed_duration

        if interrupted:
            queued_duration = self._audio_source.queued_duration
            while not self._audio_buf.empty():
                queued_duration += self._audio_buf.recv_nowait().duration

            pushed_duration = max(pushed_duration - queued_duration, 0)
            self._audio_source.clear_queue()
            wait_for_playout.cancel()
        else:
            wait_for_interruption.cancel()

        self._pushed_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:
            if not self._playback_enabled.is_set():
                self._audio_source.clear_queue()
                await self._playback_enabled.wait()
                # TODO(long): save the frames in the queue and play them later
                # 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())
            await self._audio_source.capture_frame(frame)

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
    ) -> 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

        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:
                        self._writer = await self._create_text_writer()

                    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])