From 2e8e574ea56f1760e1e11bc4ae05b84e0d9052a0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 16 Jan 2026 17:09:11 -0500 Subject: [PATCH] Add UserIdleController, deprecate UserIdleProcessor --- changelog/3482.added.md | 1 + examples/foundational/17-detect-user-idle.py | 89 +++++--- .../aggregators/llm_response_universal.py | 34 ++- src/pipecat/processors/user_idle_processor.py | 13 ++ src/pipecat/turns/user_idle_controller.py | 173 ++++++++++++++ src/pipecat/turns/user_turn_processor.py | 30 +++ tests/test_user_idle_controller.py | 216 ++++++++++++++++++ 7 files changed, 524 insertions(+), 32 deletions(-) create mode 100644 changelog/3482.added.md create mode 100644 src/pipecat/turns/user_idle_controller.py create mode 100644 tests/test_user_idle_controller.py diff --git a/changelog/3482.added.md b/changelog/3482.added.md new file mode 100644 index 000000000..35522f36c --- /dev/null +++ b/changelog/3482.added.md @@ -0,0 +1 @@ +- Added `UserIdleController` for detecting user idle state, integrated into `LLMUserAggregator` and `UserTurnProcessor` via optional `user_idle_timeout` parameter. Emits `on_user_idle` event for application-level handling. Deprecated `UserIdleProcessor` in favor of the new compositional approach. diff --git a/examples/foundational/17-detect-user-idle.py b/examples/foundational/17-detect-user-idle.py index f1fe0d068..b4042c1d1 100644 --- a/examples/foundational/17-detect-user-idle.py +++ b/examples/foundational/17-detect-user-idle.py @@ -13,7 +13,13 @@ from loguru import logger from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import EndFrame, LLMMessagesAppendFrame, LLMRunFrame, TTSSpeakFrame +from pipecat.frames.frames import ( + EndFrame, + EndTaskFrame, + LLMMessagesAppendFrame, + LLMRunFrame, + TTSSpeakFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -22,7 +28,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.user_idle_processor import UserIdleProcessor +from pipecat.processors.frame_processor import FrameDirection from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService @@ -36,6 +42,43 @@ from pipecat.turns.user_turn_strategies import UserTurnStrategies load_dotenv(override=True) + +class IdleHandler: + """Helper class to manage user idle retry logic.""" + + def __init__(self): + self._retry_count = 0 + + def reset(self): + """Reset the retry count when user becomes active.""" + self._retry_count = 0 + + async def handle_idle(self, aggregator): + """Handle user idle event with escalating prompts.""" + self._retry_count += 1 + + if self._retry_count == 1: + # First attempt: Add a gentle prompt to the conversation + message = { + "role": "system", + "content": "The user has been quiet. Politely and briefly ask if they're still there.", + } + await aggregator.push_frame(LLMMessagesAppendFrame([message], run_llm=True)) + elif self._retry_count == 2: + # Second attempt: More direct prompt + message = { + "role": "system", + "content": "The user is still inactive. Ask if they'd like to continue our conversation.", + } + await aggregator.push_frame(LLMMessagesAppendFrame([message], run_llm=True)) + else: + # Third attempt: End the conversation + await aggregator.push_frame( + TTSSpeakFrame("It seems like you're busy right now. Have a nice day!") + ) + await aggregator.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -84,42 +127,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): user_turn_strategies=UserTurnStrategies( stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())] ), + user_idle_timeout=5.0, # Detect user idle after 5 seconds ), ) - async def handle_user_idle(user_idle: UserIdleProcessor, retry_count: int) -> bool: - if retry_count == 1: - # First attempt: Add a gentle prompt to the conversation - message = { - "role": "system", - "content": "The user has been quiet. Politely and briefly ask if they're still there.", - } - await user_idle.push_frame(LLMMessagesAppendFrame([message], run_llm=True)) - return True - elif retry_count == 2: - # Second attempt: More direct prompt - message = { - "role": "system", - "content": "The user is still inactive. Ask if they'd like to continue our conversation.", - } - await user_idle.push_frame(LLMMessagesAppendFrame([message], run_llm=True)) - return True - else: - # Third attempt: End the conversation - await user_idle.push_frame( - TTSSpeakFrame("It seems like you're busy right now. Have a nice day!") - ) - await task.queue_frame(EndFrame()) - return False - - user_idle = UserIdleProcessor(callback=handle_user_idle, timeout=5.0) - pipeline = Pipeline( [ transport.input(), # Transport user input stt, - user_idle, # Idle user check-in - user_aggregator, + user_aggregator, # User aggregator with built-in idle detection llm, # LLM tts, # TTS transport.output(), # Transport bot output @@ -136,6 +152,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, ) + # Set up idle handling with retry logic + idle_handler = IdleHandler() + + @user_aggregator.event_handler("on_user_idle") + async def handle_user_idle(aggregator): + await idle_handler.handle_idle(aggregator) + + @user_aggregator.event_handler("on_user_turn_started") + async def handle_user_turn_started(aggregator, strategy): + idle_handler.reset() + @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 26c388fe6..4b0e89008 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -62,6 +62,7 @@ from pipecat.processors.aggregators.llm_context import ( NotGiven, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_mute import BaseUserMuteStrategy from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams @@ -80,11 +81,16 @@ class LLMUserAggregatorParams: user_mute_strategies: List of user mute strategies. user_turn_stop_timeout: Time in seconds to wait before considering the user's turn finished. + user_idle_timeout: Optional timeout in seconds for detecting user idle state. + If set, the aggregator will emit an `on_user_idle` event when the user + has been idle (not speaking) for this duration. Set to None to disable + idle detection. """ user_turn_strategies: Optional[UserTurnStrategies] = None user_mute_strategies: List[BaseUserMuteStrategy] = field(default_factory=list) user_turn_stop_timeout: float = 5.0 + user_idle_timeout: Optional[float] = None @dataclass @@ -291,11 +297,12 @@ class LLMUserAggregator(LLMContextAggregator): - on_user_turn_started: Called when the user turn starts - on_user_turn_stopped: Called when the user turn ends - on_user_turn_stop_timeout: Called when no user turn stop strategy triggers + - on_user_idle: Called when the user has been idle for the configured timeout Example:: @aggregator.event_handler("on_user_turn_started") - async def on_user_turn_started(aggregator, strategy: BaseUserTurnStartStrategy]): + async def on_user_turn_started(aggregator, strategy: BaseUserTurnStartStrategy): ... @aggregator.event_handler("on_user_turn_stopped") @@ -306,6 +313,10 @@ class LLMUserAggregator(LLMContextAggregator): async def on_user_turn_stop_timeout(aggregator): ... + @aggregator.event_handler("on_user_idle") + async def on_user_idle(aggregator): + ... + """ def __init__( @@ -328,6 +339,7 @@ class LLMUserAggregator(LLMContextAggregator): self._register_event_handler("on_user_turn_started") self._register_event_handler("on_user_turn_stopped") self._register_event_handler("on_user_turn_stop_timeout") + self._register_event_handler("on_user_idle") user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies() @@ -350,6 +362,14 @@ class LLMUserAggregator(LLMContextAggregator): "on_user_turn_stop_timeout", self._on_user_turn_stop_timeout ) + # Optional user idle controller + self._user_idle_controller: Optional[UserIdleController] = None + if self._params.user_idle_timeout: + self._user_idle_controller = UserIdleController( + user_idle_timeout=self._params.user_idle_timeout + ) + self._user_idle_controller.add_event_handler("on_user_idle", self._on_user_idle) + async def cleanup(self): """Clean up processor resources.""" await super().cleanup() @@ -405,6 +425,9 @@ class LLMUserAggregator(LLMContextAggregator): await self._user_turn_controller.process_frame(frame) + if self._user_idle_controller: + await self._user_idle_controller.process_frame(frame) + async def push_aggregation(self) -> str: """Push the current aggregation.""" if len(self._aggregation) == 0: @@ -420,6 +443,9 @@ class LLMUserAggregator(LLMContextAggregator): async def _start(self, frame: StartFrame): await self._user_turn_controller.setup(self.task_manager) + if self._user_idle_controller: + await self._user_idle_controller.setup(self.task_manager) + for s in self._params.user_mute_strategies: await s.setup(self.task_manager) @@ -432,6 +458,9 @@ class LLMUserAggregator(LLMContextAggregator): async def _cleanup(self): await self._user_turn_controller.cleanup() + if self._user_idle_controller: + await self._user_idle_controller.cleanup() + for s in self._params.user_mute_strategies: await s.cleanup() @@ -565,6 +594,9 @@ class LLMUserAggregator(LLMContextAggregator): async def _on_user_turn_stop_timeout(self, controller): await self._call_event_handler("on_user_turn_stop_timeout") + async def _on_user_idle(self, controller): + await self._call_event_handler("on_user_idle") + class LLMAssistantAggregator(LLMContextAggregator): """Assistant LLM aggregator that processes bot responses and function calls. diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 6dc6dd47c..67c41ab13 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -8,6 +8,7 @@ import asyncio import inspect +import warnings from typing import Awaitable, Callable, Union from pipecat.frames.frames import ( @@ -26,6 +27,10 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class UserIdleProcessor(FrameProcessor): """Monitors user inactivity and triggers callbacks after timeout periods. + .. deprecated:: + UserIdleProcessor is deprecated in 0.0.100 and will be removed in a future version. + Use LLMUserAggregator with user_idle_timeout parameter instead. + This processor tracks user activity and triggers configurable callbacks when users become idle. It starts monitoring only after the first conversation activity and supports both basic and retry-based callback patterns. @@ -70,6 +75,14 @@ class UserIdleProcessor(FrameProcessor): **kwargs: Additional arguments passed to FrameProcessor. """ super().__init__(**kwargs) + + warnings.warn( + "UserIdleProcessor is deprecated in 0.0.100 and will be removed in a " + "future version. Use LLMUserAggregator with user_idle_timeout parameter " + "instead.", + DeprecationWarning, + ) + self._callback = self._wrap_callback(callback) self._timeout = timeout self._retry_count = 0 diff --git a/src/pipecat/turns/user_idle_controller.py b/src/pipecat/turns/user_idle_controller.py new file mode 100644 index 000000000..4157d6d30 --- /dev/null +++ b/src/pipecat/turns/user_idle_controller.py @@ -0,0 +1,173 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""This module defines a controller for managing user idle detection.""" + +import asyncio +from typing import Optional + +from pipecat.frames.frames import ( + BotSpeakingFrame, + Frame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, + UserSpeakingFrame, + UserStartedSpeakingFrame, +) +from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.base_object import BaseObject + + +class UserIdleController(BaseObject): + """Controller for managing user idle detection. + + This class monitors user activity and triggers an event when the user has been + idle (not speaking) for a configured timeout period. It only starts monitoring + after the first conversation activity and does not trigger while the bot is + speaking or function calls are in progress. + + The controller tracks activity using continuous frames (UserSpeakingFrame and + BotSpeakingFrame) which are emitted repeatedly while speaking is happening, and + state-based tracking for function calls (FunctionCallsStartedFrame and + FunctionCallResultFrame) which are only sent at start and end. + + Event handlers available: + + - on_user_idle: Emitted when the user has been idle for the timeout period. + + Example:: + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + # Handle user idle - send reminder, prompt, etc. + ... + """ + + def __init__( + self, + *, + user_idle_timeout: float, + ): + """Initialize the user idle controller. + + Args: + user_idle_timeout: Timeout in seconds before considering the user idle. + """ + super().__init__() + + self._user_idle_timeout = user_idle_timeout + + self._task_manager: Optional[BaseTaskManager] = None + + self._conversation_started = False + self._function_call_in_progress = False + + self.user_idle_event = asyncio.Event() + self.user_idle_task: Optional[asyncio.Task] = None + + self._register_event_handler("on_user_idle", sync=True) + + @property + def task_manager(self) -> BaseTaskManager: + """Returns the configured task manager.""" + if not self._task_manager: + raise RuntimeError(f"{self} user idle controller was not properly setup") + return self._task_manager + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the controller with the given task manager. + + Args: + task_manager: The task manager to be associated with this instance. + """ + self._task_manager = task_manager + + if not self.user_idle_task: + self.user_idle_task = self.task_manager.create_task( + self.user_idle_task_handler(), + f"{self}::user_idle_task_handler", + ) + + async def cleanup(self): + """Cleanup the controller.""" + await super().cleanup() + + if self.user_idle_task: + await self.task_manager.cancel_task(self.user_idle_task) + self.user_idle_task = None + + async def process_frame(self, frame: Frame): + """Process an incoming frame to track user activity state. + + Args: + frame: The frame to be processed. + """ + # Start monitoring on first conversation activity + if not self._conversation_started: + if isinstance(frame, (UserStartedSpeakingFrame, BotSpeakingFrame)): + self._conversation_started = True + self.user_idle_event.set() + else: + return + + # Reset idle timer on continuous activity frames + if isinstance(frame, (UserSpeakingFrame, BotSpeakingFrame)): + await self._handle_activity(frame) + # Track function call state (start/end frames, not continuous) + elif isinstance(frame, FunctionCallsStartedFrame): + await self._handle_function_calls_started(frame) + elif isinstance(frame, FunctionCallResultFrame): + await self._handle_function_call_result(frame) + + async def _handle_activity(self, _: UserSpeakingFrame | BotSpeakingFrame): + """Handle continuous activity frames that should reset the idle timer. + + These frames are emitted continuously while the user or bot is speaking, + so we simply reset the timer whenever we receive them. + + Args: + frame: The activity frame to process. + """ + self.user_idle_event.set() + + async def _handle_function_calls_started(self, _: FunctionCallsStartedFrame): + """Handle function calls started event. + + Function calls can take longer than the timeout, so we track their state + to prevent idle callbacks while they're in progress. + + Args: + frame: The FunctionCallsStartedFrame to process. + """ + self._function_call_in_progress = True + self.user_idle_event.set() + + async def _handle_function_call_result(self, _: FunctionCallResultFrame): + """Handle function call result event. + + Args: + frame: The FunctionCallResultFrame to process. + """ + self._function_call_in_progress = False + self.user_idle_event.set() + + async def user_idle_task_handler(self): + """Monitors for idle timeout and triggers events. + + Runs in a loop until cancelled. The idle timer is reset whenever activity + frames are received (UserSpeakingFrame or BotSpeakingFrame). Function calls + are tracked via state since they only send start/end frames. If no activity + is detected for the configured timeout period and no function call is in + progress, the on_user_idle event is triggered. + """ + while True: + try: + await asyncio.wait_for(self.user_idle_event.wait(), timeout=self._user_idle_timeout) + self.user_idle_event.clear() + except asyncio.TimeoutError: + # Only trigger if conversation has started and no function call is in progress + if self._conversation_started and not self._function_call_in_progress: + await self._call_event_handler("on_user_idle") diff --git a/src/pipecat/turns/user_turn_processor.py b/src/pipecat/turns/user_turn_processor.py index 79e4664a0..62693c07d 100644 --- a/src/pipecat/turns/user_turn_processor.py +++ b/src/pipecat/turns/user_turn_processor.py @@ -19,6 +19,7 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams from pipecat.turns.user_turn_controller import UserTurnController @@ -38,6 +39,7 @@ class UserTurnProcessor(FrameProcessor): - on_user_turn_started: Emitted when a user turn starts. - on_user_turn_stopped: Emitted when a user turn stops. - on_user_turn_stop_timeout: Emitted if no stop strategy triggers before timeout. + - on_user_idle: Emitted when the user has been idle for the configured timeout. Example:: @@ -53,6 +55,10 @@ class UserTurnProcessor(FrameProcessor): async def on_user_turn_stop_timeout(processor): ... + @processor.event_handler("on_user_idle") + async def on_user_idle(processor): + ... + """ def __init__( @@ -60,6 +66,7 @@ class UserTurnProcessor(FrameProcessor): *, user_turn_strategies: Optional[UserTurnStrategies] = None, user_turn_stop_timeout: float = 5.0, + user_idle_timeout: Optional[float] = None, **kwargs, ): """Initialize the user turn processor. @@ -68,6 +75,10 @@ class UserTurnProcessor(FrameProcessor): user_turn_strategies: Configured strategies for starting and stopping user turns. user_turn_stop_timeout: Timeout in seconds to automatically stop a user turn if no activity is detected. + user_idle_timeout: Optional timeout in seconds for detecting user idle state. + If set, the processor will emit an `on_user_idle` event when the user + has been idle (not speaking) for this duration. Set to None to disable + idle detection. **kwargs: Additional keyword arguments. """ super().__init__(**kwargs) @@ -75,6 +86,7 @@ class UserTurnProcessor(FrameProcessor): self._register_event_handler("on_user_turn_started") self._register_event_handler("on_user_turn_stopped") self._register_event_handler("on_user_turn_stop_timeout") + self._register_event_handler("on_user_idle") self._user_turn_controller = UserTurnController( user_turn_strategies=user_turn_strategies or UserTurnStrategies(), @@ -92,6 +104,12 @@ class UserTurnProcessor(FrameProcessor): "on_user_turn_stop_timeout", self._on_user_turn_stop_timeout ) + # Optional user idle controller + self._user_idle_controller: Optional[UserIdleController] = None + if user_idle_timeout: + self._user_idle_controller = UserIdleController(user_idle_timeout=user_idle_timeout) + self._user_idle_controller.add_event_handler("on_user_idle", self._on_user_idle) + async def cleanup(self): """Clean up processor resources.""" await super().cleanup() @@ -129,9 +147,15 @@ class UserTurnProcessor(FrameProcessor): await self._user_turn_controller.process_frame(frame) + if self._user_idle_controller: + await self._user_idle_controller.process_frame(frame) + async def _start(self, frame: StartFrame): await self._user_turn_controller.setup(self.task_manager) + if self._user_idle_controller: + await self._user_idle_controller.setup(self.task_manager) + async def _stop(self, frame: EndFrame): await self._cleanup() @@ -141,6 +165,9 @@ class UserTurnProcessor(FrameProcessor): async def _cleanup(self): await self._user_turn_controller.cleanup() + if self._user_idle_controller: + await self._user_idle_controller.cleanup() + async def _on_push_frame( self, controller, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM ): @@ -180,3 +207,6 @@ class UserTurnProcessor(FrameProcessor): async def _on_user_turn_stop_timeout(self, controller): await self._call_event_handler("on_user_turn_stop_timeout") + + async def _on_user_idle(self, controller): + await self._call_event_handler("on_user_idle") diff --git a/tests/test_user_idle_controller.py b/tests/test_user_idle_controller.py new file mode 100644 index 000000000..568819eb8 --- /dev/null +++ b/tests/test_user_idle_controller.py @@ -0,0 +1,216 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest + +from pipecat.frames.frames import ( + BotSpeakingFrame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, + UserSpeakingFrame, + UserStartedSpeakingFrame, +) +from pipecat.turns.user_idle_controller import UserIdleController +from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams + +USER_IDLE_TIMEOUT = 0.2 + + +class TestUserIdleController(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.task_manager = TaskManager() + self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop())) + + async def test_basic_idle_detection(self): + """Test that idle event is triggered after timeout when no activity.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Start conversation + await controller.process_frame(UserStartedSpeakingFrame()) + + # Wait for idle timeout + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + + self.assertTrue(idle_triggered) + + await controller.cleanup() + + async def test_user_speaking_resets_idle_timer(self): + """Test that continuous UserSpeakingFrame frames reset the idle timer.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Start conversation + await controller.process_frame(UserStartedSpeakingFrame()) + + # Send UserSpeakingFrame continuously to reset timer + for _ in range(5): + await asyncio.sleep(USER_IDLE_TIMEOUT * 0.5) # 50% of timeout period + await controller.process_frame(UserSpeakingFrame()) + + self.assertFalse(idle_triggered) + + await controller.cleanup() + + async def test_bot_speaking_resets_idle_timer(self): + """Test that BotSpeakingFrame frames reset the idle timer.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Start conversation + await controller.process_frame(UserStartedSpeakingFrame()) + + # Bot speaking should reset timer + for _ in range(5): + await asyncio.sleep(USER_IDLE_TIMEOUT * 0.6) # 60% of timeout + await controller.process_frame(BotSpeakingFrame()) + + self.assertFalse(idle_triggered) + + await controller.cleanup() + + async def test_function_call_prevents_idle(self): + """Test that function calls in progress prevent idle event.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Start conversation + await controller.process_frame(UserStartedSpeakingFrame()) + + # Start function call + await controller.process_frame(FunctionCallsStartedFrame(function_calls=[])) + + # Wait longer than idle timeout + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + + # Should not trigger idle because function call is in progress + self.assertFalse(idle_triggered) + + # Complete function call + await controller.process_frame( + FunctionCallResultFrame( + function_name="test", + tool_call_id="123", + arguments={}, + result=None, + run_llm=False, + ) + ) + + # Now idle should trigger + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + self.assertTrue(idle_triggered) + + await controller.cleanup() + + async def test_no_idle_before_conversation_starts(self): + """Test that idle monitoring doesn't start before first conversation activity.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Wait without starting conversation + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + + self.assertFalse(idle_triggered) + + await controller.cleanup() + + async def test_idle_starts_with_bot_speaking(self): + """Test that monitoring starts with BotSpeakingFrame, not just user speech.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_triggered = False + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_triggered + idle_triggered = True + + # Start conversation with bot speaking + await controller.process_frame(BotSpeakingFrame()) + + # Wait for idle timeout + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + + self.assertTrue(idle_triggered) + + await controller.cleanup() + + async def test_multiple_idle_events(self): + """Test that idle event can trigger multiple times.""" + controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT) + await controller.setup(self.task_manager) + + idle_count = 0 + + @controller.event_handler("on_user_idle") + async def on_user_idle(controller): + nonlocal idle_count + idle_count += 1 + + # Start conversation + await controller.process_frame(UserStartedSpeakingFrame()) + + # First idle + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + first_count = idle_count + self.assertGreaterEqual(first_count, 1) + + # Second idle + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + second_count = idle_count + self.assertGreater(second_count, first_count) + + # User activity resets timer + await controller.process_frame(UserSpeakingFrame()) + + # Give a moment for the timer to reset + await asyncio.sleep(0.1) + + # Third idle + await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1) + third_count = idle_count + self.assertGreater(third_count, second_count) + + await controller.cleanup()