async def transcription_node(
        self, text: AsyncIterable[str | TimedString], model_settings: ModelSettings
    ) -> AsyncGenerator[str | TimedString, None]:
        """
        Intercepts agent transcription to stream chunks via Redis.

        IMPORTANT: To avoid orphaned streams when user interrupts before audio plays,
        we only start streaming once audio has actually started (agent_state == "speaking").
        
        This follows the same pattern as LiveKit's TranscriptSynchronizer which gates
        text output based on audio timing.

        The final message is handled separately by conversation_item_added event,
        which calls session_recorder.log_and_stream_conversation_item.
        Message ID continuity is maintained via transcription_drafts.
        """
        session_recorder = getattr(self.session.userdata, "session_recorder", None)
        
        # Track streaming state
        stream_started = False
        buffered_chunks: list[str | TimedString] = []
        
        # Small polling interval to check for audio start
        AUDIO_CHECK_INTERVAL = 0.05  # 50ms
        
        async def wait_for_audio_or_interruption() -> bool:
            """
            Wait until audio starts playing (agent_state == 'speaking') or speech is interrupted.
            Returns True if audio started, False if interrupted before audio.
            """
            while True:
                # Check if audio has started
                if self.session.agent_state == "speaking":
                    return True
                
                # Check if speech was interrupted before audio started
                speech_handle: SpeechHandle | None = getattr(self.session, "current_speech", None)
                if speech_handle and speech_handle.interrupted:
                    <http://logger.info|logger.info>("[TRANSCRIPTION] Speech interrupted before audio started - discarding buffer")
                    return False
                
                # Brief wait before checking again
                await asyncio.sleep(AUDIO_CHECK_INTERVAL)
        
        async def start_stream_and_flush_buffer():
            """Start the transcription stream and flush all buffered chunks."""
            nonlocal stream_started
            if stream_started or not session_recorder:
                return
            
            stream_started = True
            try:
                await session_recorder.start_agent_transcription_stream()
                <http://logger.info|logger.info>(
                    "[TRANSCRIPTION] Audio started - flushing %d buffered chunks",
                    len(buffered_chunks)
                )

                # Flush all buffered chunks
                for buffered_delta in buffered_chunks:
                    if buffered_delta:
                        await session_recorder.publish_agent_transcription_chunk(buffered_delta)

            except Exception as e:
                logger.warning("Failed to start transcription stream: %s", e)
        
        async for delta in text:
            # Check current speech handle for interruption
            speech_handle: SpeechHandle | None = getattr(self.session, "current_speech", None)
            
            if not stream_started:
                # Audio hasn't started yet - buffer the chunk
                if delta:
                    buffered_chunks.append(delta)
                
                # Check if audio has started (agent is now speaking)
                if self.session.agent_state == "speaking":
                    await start_stream_and_flush_buffer()
                elif speech_handle and speech_handle.interrupted:
                    # Interrupted before audio started - discard buffer and stop
                    <http://logger.info|logger.info>(
                        "[TRANSCRIPTION] Interrupted before audio - discarding %d chunks",
                        len(buffered_chunks)
                    )
                    buffered_chunks.clear()
                    # Still yield the delta to maintain the pipeline
                    yield delta
                    continue
            else:
                # Stream is active - publish chunk directly
                if session_recorder and delta:
                    try:
                        await session_recorder.publish_agent_transcription_chunk(delta)
                    except Exception as e:
                        logger.warning("Failed to publish transcription chunk: %s", e)
            
            yield delta
        
        # After all chunks processed, check if we need to start stream
        # (handles edge case where audio starts after last chunk arrives)
        if not stream_started and buffered_chunks:
            # Give a brief moment for audio to start
            audio_started = await asyncio.wait_for(
                wait_for_audio_or_interruption(),
                timeout=0.5  # Max 500ms wait
            ) if buffered_chunks else False
            
            if audio_started:
                await start_stream_and_flush_buffer()
            else:
                <http://logger.info|logger.info>(
                    "[TRANSCRIPTION] Stream ended without audio start - "
                    "discarding %d buffered chunks (no orphan created)",
                    len(buffered_chunks)
                )

SpeechHandle

agent_state

current_speech.interrupted

current_speech

agent_state == "speaking"

agent_state_changed

agent_state_changed