Remove deprecated transcript_processor module

2026-04-02 10:54:13 -04:00
parent 87e8ed109a
commit 2a118084bd
2 changed files with 0 additions and 1168 deletions
--- a/src/pipecat/processors/transcript_processor.py
+++ b/src/pipecat/processors/transcript_processor.py
@@ -1,370 +0,0 @@
 #
 # Copyright (c) 2024-2026, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 """Transcript processing utilities for conversation recording and analysis.
 This module provides processors that convert speech and text frames into structured
 transcript messages with timestamps, enabling conversation history tracking and analysis.
 """
 from typing import List, Optional
 from loguru import logger
 from pipecat.frames.frames import (
    BotStoppedSpeakingFrame,
    CancelFrame,
    EndFrame,
    Frame,
    InterruptionFrame,
    LLMThoughtEndFrame,
    LLMThoughtStartFrame,
    LLMThoughtTextFrame,
    ThoughtTranscriptionMessage,
    TranscriptionFrame,
    TranscriptionMessage,
    TranscriptionUpdateFrame,
    TTSTextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
 from pipecat.utils.time import time_now_iso8601
 class BaseTranscriptProcessor(FrameProcessor):
    """Base class for processing conversation transcripts.
    Provides common functionality for handling transcript messages and updates.
    """
    def __init__(self, **kwargs):
        """Initialize processor with empty message store.
        Args:
            **kwargs: Additional arguments passed to parent class.
        """
        super().__init__(**kwargs)
        self._processed_messages: List[TranscriptionMessage] = []
        self._register_event_handler("on_transcript_update")
    async def _emit_update(self, messages: List[TranscriptionMessage]):
        """Emit transcript updates for new messages.
        Args:
            messages: New messages to emit in update.
        """
        if messages:
            self._processed_messages.extend(messages)
            update_frame = TranscriptionUpdateFrame(messages=messages)
            await self._call_event_handler("on_transcript_update", update_frame)
            await self.push_frame(update_frame)
 class UserTranscriptProcessor(BaseTranscriptProcessor):
    """Processes user transcription frames into timestamped conversation messages."""
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Process TranscriptionFrames into user conversation messages.
        Args:
            frame: Input frame to process.
            direction: Frame processing direction.
        """
        await super().process_frame(frame, direction)
        if isinstance(frame, TranscriptionFrame):
            message = TranscriptionMessage(
                role="user", user_id=frame.user_id, content=frame.text, timestamp=frame.timestamp
            )
            await self._emit_update([message])
        await self.push_frame(frame, direction)
 class AssistantTranscriptProcessor(BaseTranscriptProcessor):
    """Processes assistant TTS text frames and LLM thought frames into timestamped messages.
    This processor aggregates both TTS text frames and LLM thought frames into
    complete utterances and thoughts, emitting them as transcript messages.
    An assistant utterance is completed when:
    - The bot stops speaking (BotStoppedSpeakingFrame)
    - The bot is interrupted (InterruptionFrame)
    - The pipeline ends (EndFrame, CancelFrame)
    A thought is completed when:
    - The thought ends (LLMThoughtEndFrame)
    - The bot is interrupted (InterruptionFrame)
    - The pipeline ends (EndFrame, CancelFrame)
    """
    def __init__(self, *, process_thoughts: bool = False, **kwargs):
        """Initialize processor with aggregation state.
        Args:
            process_thoughts: Whether to process LLM thought frames. Defaults to False.
            **kwargs: Additional arguments passed to parent class.
        """
        super().__init__(**kwargs)
        self._process_thoughts = process_thoughts
        self._current_assistant_text_parts: List[TextPartForConcatenation] = []
        self._assistant_text_start_time: Optional[str] = None
        self._current_thought_parts: List[TextPartForConcatenation] = []
        self._thought_start_time: Optional[str] = None
        self._thought_active = False
    async def _emit_aggregated_assistant_text(self):
        """Aggregates and emits text fragments as a transcript message.
        This method aggregates text fragments that may arrive in multiple
        TTSTextFrame instances and emits them as a single TranscriptionMessage.
        """
        if self._current_assistant_text_parts and self._assistant_text_start_time:
            content = concatenate_aggregated_text(self._current_assistant_text_parts)
            if content:
                logger.trace(f"Emitting aggregated assistant message: {content}")
                message = TranscriptionMessage(
                    role="assistant",
                    content=content,
                    timestamp=self._assistant_text_start_time,
                )
                await self._emit_update([message])
            else:
                logger.trace("No content to emit after stripping whitespace")
            # Reset aggregation state
            self._current_assistant_text_parts = []
            self._assistant_text_start_time = None
    async def _emit_aggregated_thought(self):
        """Aggregates and emits thought text fragments as a thought transcript message.
        This method aggregates thought fragments that may arrive in multiple
        LLMThoughtTextFrame instances and emits them as a single ThoughtTranscriptionMessage.
        """
        if self._current_thought_parts and self._thought_start_time:
            content = concatenate_aggregated_text(self._current_thought_parts)
            if content:
                logger.trace(f"Emitting aggregated thought message: {content}")
                message = ThoughtTranscriptionMessage(
                    content=content,
                    timestamp=self._thought_start_time,
                )
                await self._emit_update([message])
            else:
                logger.trace("No thought content to emit after stripping whitespace")
            # Reset aggregation state
            self._current_thought_parts = []
            self._thought_start_time = None
            self._thought_active = False
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        """Process frames into assistant conversation messages and thought messages.
        Handles different frame types:
        - TTSTextFrame: Aggregates text for current utterance
        - LLMThoughtStartFrame: Begins aggregating a new thought
        - LLMThoughtTextFrame: Aggregates text for current thought
        - LLMThoughtEndFrame: Completes current thought
        - BotStoppedSpeakingFrame: Completes current utterance
        - InterruptionFrame: Completes current utterance and thought due to interruption
        - EndFrame: Completes current utterance and thought at pipeline end
        - CancelFrame: Completes current utterance and thought due to cancellation
        Args:
            frame: Input frame to process.
            direction: Frame processing direction.
        """
        await super().process_frame(frame, direction)
        if isinstance(frame, (InterruptionFrame, CancelFrame)):
            # Push frame first otherwise our emitted transcription update frame
            # might get cleaned up.
            await self.push_frame(frame, direction)
            # Emit accumulated text and thought with interruptions
            await self._emit_aggregated_assistant_text()
            if self._process_thoughts and self._thought_active:
                await self._emit_aggregated_thought()
        elif isinstance(frame, LLMThoughtStartFrame):
            # Start a new thought
            if self._process_thoughts:
                self._thought_active = True
                self._thought_start_time = time_now_iso8601()
                self._current_thought_parts = []
            # Push frame.
            await self.push_frame(frame, direction)
        elif isinstance(frame, LLMThoughtTextFrame):
            # Aggregate thought text if we have an active thought
            if self._process_thoughts and self._thought_active:
                self._current_thought_parts.append(
                    TextPartForConcatenation(
                        frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
                    )
                )
            # Push frame.
            await self.push_frame(frame, direction)
        elif isinstance(frame, LLMThoughtEndFrame):
            # Emit accumulated thought when thought ends
            if self._process_thoughts and self._thought_active:
                await self._emit_aggregated_thought()
            # Push frame.
            await self.push_frame(frame, direction)
        elif isinstance(frame, TTSTextFrame):
            # Start timestamp on first text part
            if not self._assistant_text_start_time:
                self._assistant_text_start_time = time_now_iso8601()
            self._current_assistant_text_parts.append(
                TextPartForConcatenation(
                    frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
                )
            )
            # Push frame.
            await self.push_frame(frame, direction)
        elif isinstance(frame, (BotStoppedSpeakingFrame, EndFrame)):
            # Emit accumulated text when bot finishes speaking or pipeline ends.
            await self._emit_aggregated_assistant_text()
            # Emit accumulated thought at pipeline end if still active
            if isinstance(frame, EndFrame) and self._process_thoughts and self._thought_active:
                await self._emit_aggregated_thought()
            # Push frame.
            await self.push_frame(frame, direction)
        else:
            await self.push_frame(frame, direction)
 class TranscriptProcessor:
    """Factory for creating and managing transcript processors.
    Provides unified access to user and assistant transcript processors
    with shared event handling. The assistant processor handles both TTS text
    and LLM thought frames.
    Example::
        transcript = TranscriptProcessor()
        pipeline = Pipeline(
            [
                transport.input(),
                stt,
                transcript.user(),              # User transcripts
                context_aggregator.user(),
                llm,
                tts,
                transport.output(),
                transcript.assistant(),         # Assistant transcripts (including thoughts)
                context_aggregator.assistant(),
            ]
        )
        @transcript.event_handler("on_transcript_update")
        async def handle_update(processor, frame):
            print(f"New messages: {frame.messages}")
    .. deprecated:: 0.0.99
        `TranscriptProcessor` is deprecated and will be removed in a future version.
        Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.
    """
    def __init__(self, *, process_thoughts: bool = False):
        """Initialize factory.
        Args:
            process_thoughts: Whether the assistant processor should handle LLM thought
                frames. Defaults to False.
        """
        self._process_thoughts = process_thoughts
        self._user_processor = None
        self._assistant_processor = None
        self._event_handlers = {}
        import warnings
        with warnings.catch_warnings():
            warnings.simplefilter("always")
            warnings.warn(
                "`TranscriptProcessor` is deprecated and will be removed in a future version. "
                "Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.",
                DeprecationWarning,
            )
    def user(self, **kwargs) -> UserTranscriptProcessor:
        """Get the user transcript processor.
        Args:
            **kwargs: Arguments specific to UserTranscriptProcessor.
        Returns:
            The user transcript processor instance.
        """
        if self._user_processor is None:
            self._user_processor = UserTranscriptProcessor(**kwargs)
            # Apply any registered event handlers
            for event_name, handler in self._event_handlers.items():
                @self._user_processor.event_handler(event_name)
                async def user_handler(processor, frame):
                    return await handler(processor, frame)
        return self._user_processor
    def assistant(self, **kwargs) -> AssistantTranscriptProcessor:
        """Get the assistant transcript processor.
        Args:
            **kwargs: Arguments specific to AssistantTranscriptProcessor.
        Returns:
            The assistant transcript processor instance.
        """
        if self._assistant_processor is None:
            self._assistant_processor = AssistantTranscriptProcessor(
                process_thoughts=self._process_thoughts, **kwargs
            )
            # Apply any registered event handlers
            for event_name, handler in self._event_handlers.items():
                @self._assistant_processor.event_handler(event_name)
                async def assistant_handler(processor, frame):
                    return await handler(processor, frame)
        return self._assistant_processor
    def event_handler(self, event_name: str):
        """Register event handler for both processors.
        Args:
            event_name: Name of event to handle.
        Returns:
            Decorator function that registers handler with both processors.
        """
        def decorator(handler):
            self._event_handlers[event_name] = handler
            # Apply handler to existing processors if they exist
            if self._user_processor:
                @self._user_processor.event_handler(event_name)
                async def user_handler(processor, frame):
                    return await handler(processor, frame)
            if self._assistant_processor:
                @self._assistant_processor.event_handler(event_name)
                async def assistant_handler(processor, frame):
                    return await handler(processor, frame)
            return handler
        return decorator
--- a/tests/test_transcript_processor.py
+++ b/tests/test_transcript_processor.py
@@ -1,798 +0,0 @@
 #
 # Copyright (c) 2024-2026, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 import asyncio
 import unittest
 from datetime import datetime, timezone
 from typing import List, Tuple, cast
 from pipecat.frames.frames import (
    AggregationType,
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    CancelFrame,
    InterruptionFrame,
    LLMThoughtEndFrame,
    LLMThoughtStartFrame,
    LLMThoughtTextFrame,
    ThoughtTranscriptionMessage,
    TranscriptionFrame,
    TranscriptionMessage,
    TranscriptionUpdateFrame,
    TTSTextFrame,
 )
 from pipecat.processors.transcript_processor import (
    AssistantTranscriptProcessor,
    UserTranscriptProcessor,
 )
 from pipecat.tests.utils import SleepFrame, run_test
 class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
    """Tests for UserTranscriptProcessor"""
    async def test_basic_transcription(self):
        """Test basic transcription frame processing"""
        # Create processor
        processor = UserTranscriptProcessor()
        # Create test timestamp
        timestamp = datetime.now(timezone.utc).isoformat()
        # Create frames to send
        frames_to_send = [
            TranscriptionFrame(text="Hello, world!", user_id="test_user", timestamp=timestamp)
        ]
        # Expected frames downstream - note the order:
        # 1. TranscriptionUpdateFrame (processor emits the update first)
        # 2. TranscriptionFrame (original frame is passed through)
        expected_down_frames = [TranscriptionUpdateFrame, TranscriptionFrame]
        # Run test
        received_frames, _ = await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify the content of the TranscriptionUpdateFrame
        update_frame = cast(
            TranscriptionUpdateFrame, received_frames[0]
        )  # Note: now checking first frame
        self.assertIsInstance(update_frame, TranscriptionUpdateFrame)
        self.assertEqual(len(update_frame.messages), 1)
        message = update_frame.messages[0]
        self.assertEqual(message.role, "user")
        self.assertEqual(message.content, "Hello, world!")
        self.assertEqual(message.user_id, "test_user")
        self.assertEqual(message.timestamp, timestamp)
    async def test_event_handler(self):
        """Test that event handlers are called with transcript updates"""
        # Create processor
        processor = UserTranscriptProcessor()
        # Track received updates
        received_updates: List[TranscriptionMessage] = []
        # Register event handler
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.extend(frame.messages)
        # Create test data
        timestamp = datetime.now(timezone.utc).isoformat()
        frames_to_send = [
            TranscriptionFrame(text="First message", user_id="test_user", timestamp=timestamp),
            TranscriptionFrame(text="Second message", user_id="test_user", timestamp=timestamp),
        ]
        expected_down_frames = [
            TranscriptionUpdateFrame,
            TranscriptionFrame,  # First message
            TranscriptionUpdateFrame,
            TranscriptionFrame,  # Second message
        ]
        # Run test
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify event handler received updates
        self.assertEqual(len(received_updates), 2)
        # Check first message
        self.assertEqual(received_updates[0].role, "user")
        self.assertEqual(received_updates[0].content, "First message")
        self.assertEqual(received_updates[0].timestamp, timestamp)
        # Check second message
        self.assertEqual(received_updates[1].role, "user")
        self.assertEqual(received_updates[1].content, "Second message")
        self.assertEqual(received_updates[1].timestamp, timestamp)
    async def test_text_aggregation(self):
        """Test that TTSTextFrames are properly aggregated into a single message"""
        # Create processor
        processor = AssistantTranscriptProcessor()
        # Track received updates
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Create test frames simulating bot speaking multiple text chunks
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),  # Wait for StartedSpeaking to process
            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="How", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="are", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="you?", aggregated_by=AggregationType.WORD),
            SleepFrame(),  # Wait for text frames to queue
            BotStoppedSpeakingFrame(),
        ]
        # Expected order:
        # 1. BotStartedSpeakingFrame (system frame, immediate)
        # 2. All queued TTSTextFrames
        # 3. BotStoppedSpeakingFrame (system frame, immediate)
        # 4. TranscriptionUpdateFrame (after aggregation)
        expected_down_frames = [
            BotStartedSpeakingFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TranscriptionUpdateFrame,
            BotStoppedSpeakingFrame,
        ]
        # Run test
        received_frames, _ = await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify update was received
        self.assertEqual(len(received_updates), 1)
        # Get the update frame
        update_frame = received_updates[0]
        # Should have one aggregated message
        self.assertEqual(len(update_frame.messages), 1)
        message = update_frame.messages[0]
        self.assertEqual(message.role, "assistant")
        self.assertEqual(message.content, "Hello world! How are you?")
        # Verify timestamp exists
        self.assertIsNotNone(message.timestamp)
        # All frames should be passed through in order, with update at end
        downstream_update = cast(TranscriptionUpdateFrame, received_frames[-2])
        self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?")
    async def test_empty_text_handling(self):
        """Test that empty messages are not emitted"""
        processor = AssistantTranscriptProcessor()
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            TTSTextFrame(text="", aggregated_by=AggregationType.WORD),  # Empty text
            TTSTextFrame(text="   ", aggregated_by=AggregationType.WORD),  # Just whitespace
            TTSTextFrame(text="\n", aggregated_by=AggregationType.WORD),  # Just newline
            BotStoppedSpeakingFrame(),
            # Pipeline ends here; run_test will automatically send EndFrame
        ]
        # From our earlier tests, we know BotStoppedSpeakingFrame comes before TTSTextFrames
        expected_down_frames = [
            BotStartedSpeakingFrame,
            BotStoppedSpeakingFrame,
            TTSTextFrame,  # empty
            TTSTextFrame,  # whitespace
            TTSTextFrame,  # newline
            # No TranscriptionUpdateFrame since content is empty after stripping
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        self.assertEqual(len(received_updates), 0, "No updates should be emitted for empty content")
    async def test_interruption_handling(self):
        """Test that messages are properly captured when bot is interrupted"""
        processor = AssistantTranscriptProcessor()
        # Track received updates
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Simulate bot being interrupted mid-sentence
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
            SleepFrame(),
            InterruptionFrame(),  # User interrupts here
            SleepFrame(),
            BotStartedSpeakingFrame(),
            TTSTextFrame(text="New", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="response", aggregated_by=AggregationType.WORD),
            SleepFrame(),
            BotStoppedSpeakingFrame(),
        ]
        # Actual order of frames:
        expected_down_frames = [
            BotStartedSpeakingFrame,
            TTSTextFrame,  # "Hello"
            TTSTextFrame,  # "world!"
            InterruptionFrame,
            TranscriptionUpdateFrame,  # First message (emitted due to interruption)
            BotStartedSpeakingFrame,
            TTSTextFrame,  # "New"
            TTSTextFrame,  # "response"
            TranscriptionUpdateFrame,  # Second message
            BotStoppedSpeakingFrame,
        ]
        # Run test
        received_frames, _ = await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Should have received two updates
        self.assertEqual(len(received_updates), 2)
        # First update should be interrupted message
        first_message = received_updates[0].messages[0]
        self.assertEqual(first_message.role, "assistant")
        self.assertEqual(first_message.content, "Hello world!")
        self.assertIsNotNone(first_message.timestamp)
        # Second update should be new response
        second_message = received_updates[1].messages[0]
        self.assertEqual(second_message.role, "assistant")
        self.assertEqual(second_message.content, "New response")
        self.assertIsNotNone(second_message.timestamp)
        # Verify timestamps are different
        self.assertNotEqual(first_message.timestamp, second_message.timestamp)
    async def test_end_frame_handling(self):
        """Test that final messages are captured when pipeline ends normally"""
        processor = AssistantTranscriptProcessor()
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
            # Pipeline ends here; run_test will automatically send EndFrame
        ]
        expected_down_frames = [
            BotStartedSpeakingFrame,
            TTSTextFrame,
            TTSTextFrame,
            TranscriptionUpdateFrame,  # Final message emitted due to EndFrame
        ]
        # Run test - EndFrame will be sent automatically
        received_frames, _ = await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertEqual(message.role, "assistant")
        self.assertEqual(message.content, "Hello world")
    async def test_cancel_frame_handling(self):
        """Test that messages are properly captured when pipeline is cancelled"""
        processor = AssistantTranscriptProcessor()
        # Track updates with timestamps to verify order
        received_updates: List[Tuple[str, float]] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            # Record message content and time received
            received_updates.append((frame.messages[0].content, asyncio.get_event_loop().time()))
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
            SleepFrame(),  # Ensure messages are processed
            CancelFrame(),
        ]
        # We don't need to verify frame order, just that CancelFrame triggers message emission
        expected_down_frames = [
            BotStartedSpeakingFrame,
            TTSTextFrame,
            TTSTextFrame,
            CancelFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
            send_end_frame=False,
        )
        # Verify that we received an update
        self.assertEqual(len(received_updates), 1, "Should receive one update before cancellation")
        content, _ = received_updates[0]
        self.assertEqual(content, "Hello world")
    async def test_transcript_processor_factory(self):
        """Test that factory properly manages processors and event handlers"""
        from pipecat.processors.transcript_processor import TranscriptProcessor
        factory = TranscriptProcessor()
        received_updates: List[TranscriptionMessage] = []
        # Register handler with factory
        @factory.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.extend(frame.messages)
        # Get processors and verify they're reused
        user_proc1 = factory.user()
        user_proc2 = factory.user()
        self.assertIs(user_proc1, user_proc2, "User processor should be reused")
        asst_proc1 = factory.assistant()
        asst_proc2 = factory.assistant()
        self.assertIs(asst_proc1, asst_proc2, "Assistant processor should be reused")
        # Test user processor
        timestamp = datetime.now(timezone.utc).isoformat()
        frames_to_send = [
            TranscriptionFrame(text="User message", user_id="user1", timestamp=timestamp)
        ]
        await run_test(
            user_proc1,
            frames_to_send=frames_to_send,
            expected_down_frames=[TranscriptionUpdateFrame, TranscriptionFrame],
        )
        # Test assistant processor
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            TTSTextFrame(text="Assistant", aggregated_by=AggregationType.WORD),
            TTSTextFrame(text="message", aggregated_by=AggregationType.WORD),
            BotStoppedSpeakingFrame(),
        ]
        # The actual order we see in the output:
        await run_test(
            asst_proc1,
            frames_to_send=frames_to_send,
            expected_down_frames=[
                BotStartedSpeakingFrame,
                BotStoppedSpeakingFrame,
                TTSTextFrame,
                TTSTextFrame,
                TranscriptionUpdateFrame,
            ],
        )
        # Verify both processors triggered the same handler
        self.assertEqual(len(received_updates), 2)
        self.assertEqual(received_updates[0].role, "user")
        self.assertEqual(received_updates[0].content, "User message")
        self.assertEqual(received_updates[1].role, "assistant")
        self.assertEqual(received_updates[1].content, "Assistant message")
    async def test_text_fragments_with_spaces(self):
        """Test aggregating text fragments with various spacing patterns"""
        processor = AssistantTranscriptProcessor()
        # Track received updates
        received_updates = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Test the specific pattern shared
        def make_tts_text_frame(text: str) -> TTSTextFrame:
            frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD)
            frame.includes_inter_frame_spaces = True
            return frame
        frames_to_send = [
            BotStartedSpeakingFrame(),
            SleepFrame(),
            make_tts_text_frame("Hello"),
            make_tts_text_frame(" there"),
            make_tts_text_frame("!"),
            make_tts_text_frame(" How"),
            make_tts_text_frame("'s"),
            make_tts_text_frame(" it"),
            make_tts_text_frame(" going"),
            make_tts_text_frame("?"),
            BotStoppedSpeakingFrame(),
        ]
        expected_down_frames = [
            BotStartedSpeakingFrame,
            BotStoppedSpeakingFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TTSTextFrame,
            TranscriptionUpdateFrame,
        ]
        # Run test
        received_frames, _ = await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify result
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertEqual(message.role, "assistant")
        # Should be properly joined without extra spaces
        self.assertEqual(message.content, "Hello there! How's it going?")
 class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase):
    """Tests for thought transcription in AssistantTranscriptProcessor"""
    async def test_basic_thought_transcription(self):
        """Test basic thought frame processing"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Create frames for a simple thought
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="Let me think about this..."),
            LLMThoughtEndFrame(),
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            TranscriptionUpdateFrame,
            LLMThoughtEndFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify update was received
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertIsInstance(message, ThoughtTranscriptionMessage)
        self.assertEqual(message.content, "Let me think about this...")
        self.assertIsNotNone(message.timestamp)
    async def test_thought_aggregation(self):
        """Test that thought text frames are properly aggregated"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Create frames simulating chunked thought text
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="The user "),
            LLMThoughtTextFrame(text="is asking "),
            LLMThoughtTextFrame(text="about electric "),
            LLMThoughtTextFrame(text="cars."),
            LLMThoughtEndFrame(),
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            LLMThoughtTextFrame,
            LLMThoughtTextFrame,
            LLMThoughtTextFrame,
            TranscriptionUpdateFrame,
            LLMThoughtEndFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify aggregation
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertIsInstance(message, ThoughtTranscriptionMessage)
        self.assertEqual(message.content, "The user is asking about electric cars.")
    async def test_thought_with_interruption(self):
        """Test that thoughts are properly captured when interrupted"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="I need to consider "),
            LLMThoughtTextFrame(text="multiple factors"),
            SleepFrame(),
            InterruptionFrame(),  # User interrupts
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            LLMThoughtTextFrame,
            InterruptionFrame,
            TranscriptionUpdateFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify thought was captured on interruption
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertIsInstance(message, ThoughtTranscriptionMessage)
        self.assertEqual(message.content, "I need to consider multiple factors")
    async def test_thought_with_cancel(self):
        """Test that thoughts are properly captured when cancelled"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="Starting analysis"),
            SleepFrame(),
            CancelFrame(),
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            CancelFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
            send_end_frame=False,
        )
        # Verify thought was captured on cancellation
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertIsInstance(message, ThoughtTranscriptionMessage)
        self.assertEqual(message.content, "Starting analysis")
    async def test_thought_with_end_frame(self):
        """Test that thoughts are captured when pipeline ends normally"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="Final thought"),
            # Pipeline ends here; run_test will automatically send EndFrame
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            TranscriptionUpdateFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify thought was captured on EndFrame
        self.assertEqual(len(received_updates), 1)
        message = received_updates[0].messages[0]
        self.assertIsInstance(message, ThoughtTranscriptionMessage)
        self.assertEqual(message.content, "Final thought")
    async def test_multiple_thoughts(self):
        """Test multiple separate thoughts in sequence"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            # First thought
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="First consideration"),
            LLMThoughtEndFrame(),
            # Second thought
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text="Second consideration"),
            LLMThoughtEndFrame(),
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            TranscriptionUpdateFrame,
            LLMThoughtEndFrame,
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            TranscriptionUpdateFrame,
            LLMThoughtEndFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify both thoughts were captured
        self.assertEqual(len(received_updates), 2)
        first_message = received_updates[0].messages[0]
        self.assertIsInstance(first_message, ThoughtTranscriptionMessage)
        self.assertEqual(first_message.content, "First consideration")
        second_message = received_updates[1].messages[0]
        self.assertIsInstance(second_message, ThoughtTranscriptionMessage)
        self.assertEqual(second_message.content, "Second consideration")
    async def test_empty_thought_handling(self):
        """Test that empty thoughts are not emitted"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        frames_to_send = [
            LLMThoughtStartFrame(),
            LLMThoughtTextFrame(text=""),  # Empty
            LLMThoughtTextFrame(text="   "),  # Just whitespace
            LLMThoughtEndFrame(),
        ]
        expected_down_frames = [
            LLMThoughtStartFrame,
            LLMThoughtTextFrame,
            LLMThoughtTextFrame,
            LLMThoughtEndFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify no updates emitted for empty content
        self.assertEqual(len(received_updates), 0)
    async def test_thought_without_start_frame(self):
        """Test that thought text without start frame is ignored"""
        processor = AssistantTranscriptProcessor(process_thoughts=True)
        received_updates: List[TranscriptionUpdateFrame] = []
        @processor.event_handler("on_transcript_update")
        async def handle_update(proc, frame: TranscriptionUpdateFrame):
            received_updates.append(frame)
        # Send thought text without start frame
        frames_to_send = [
            LLMThoughtTextFrame(text="This should be ignored"),
            LLMThoughtEndFrame(),
        ]
        expected_down_frames = [
            LLMThoughtTextFrame,
            LLMThoughtEndFrame,
        ]
        await run_test(
            processor,
            frames_to_send=frames_to_send,
            expected_down_frames=expected_down_frames,
        )
        # Verify no updates since thought wasn't properly started
        self.assertEqual(len(received_updates), 0)
 if __name__ == "__main__":
    unittest.main()