Redesign UserIdleController to use BotStoppedSpeakingFrame
Replace the continuous heartbeat-based timer (UserSpeakingFrame/BotSpeakingFrame + asyncio.Event loop) with a simple one-shot timer that starts when BotStoppedSpeakingFrame is received and cancels on UserStartedSpeakingFrame or BotStartedSpeakingFrame. This eliminates false idle triggers caused by gaps between the user finishing speaking and the bot starting to speak (LLM/TTS latency). Guard the timer start with two conditions to prevent false triggers: - User turn in progress: during interruptions, BotStoppedSpeaking arrives while the user is still speaking mid-turn. - Function calls in progress: FunctionCallsStarted arrives before BotStoppedSpeaking because the bot speaks concurrently with the function call starting, so the timer must wait for the result and subsequent bot response.
This commit is contained in:
@@ -5,11 +5,14 @@
|
||||
#
|
||||
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
from pipecat.adapters.schemas.function_schema import FunctionSchema
|
||||
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
||||
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
||||
from pipecat.frames.frames import (
|
||||
EndTaskFrame,
|
||||
@@ -30,6 +33,7 @@ from pipecat.runner.types import RunnerArguments
|
||||
from pipecat.runner.utils import create_transport
|
||||
from pipecat.services.cartesia.tts import CartesiaTTSService
|
||||
from pipecat.services.deepgram.stt import DeepgramSTTService
|
||||
from pipecat.services.llm_service import FunctionCallParams
|
||||
from pipecat.services.openai.llm import OpenAILLMService
|
||||
from pipecat.transports.base_transport import BaseTransport, TransportParams
|
||||
from pipecat.transports.daily.transport import DailyParams
|
||||
@@ -74,6 +78,17 @@ class IdleHandler:
|
||||
await aggregator.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
|
||||
|
||||
|
||||
async def fetch_weather_from_api(params: FunctionCallParams):
|
||||
# Simulate a slow API call, waiting longer than the user idle timeout.
|
||||
await asyncio.sleep(3)
|
||||
await params.result_callback({"conditions": "nice", "temperature": "75"})
|
||||
|
||||
|
||||
async def fetch_restaurant_recommendation(params: FunctionCallParams):
|
||||
await asyncio.sleep(6)
|
||||
await params.result_callback({"name": "The Golden Dragon"})
|
||||
|
||||
|
||||
# We use lambdas to defer transport parameter creation until the transport
|
||||
# type is selected at runtime.
|
||||
transport_params = {
|
||||
@@ -104,6 +119,42 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
||||
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
|
||||
|
||||
llm.register_function("get_current_weather", fetch_weather_from_api)
|
||||
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
|
||||
|
||||
@llm.event_handler("on_function_calls_started")
|
||||
async def on_function_calls_started(service, function_calls):
|
||||
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
|
||||
|
||||
weather_function = FunctionSchema(
|
||||
name="get_current_weather",
|
||||
description="Get the current weather",
|
||||
properties={
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["celsius", "fahrenheit"],
|
||||
"description": "The temperature unit to use. Infer this from the user's location.",
|
||||
},
|
||||
},
|
||||
required=["location", "format"],
|
||||
)
|
||||
restaurant_function = FunctionSchema(
|
||||
name="get_restaurant_recommendation",
|
||||
description="Get a restaurant recommendation",
|
||||
properties={
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "The city and state, e.g. San Francisco, CA",
|
||||
},
|
||||
},
|
||||
required=["location"],
|
||||
)
|
||||
tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
@@ -111,7 +162,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
||||
},
|
||||
]
|
||||
|
||||
context = LLMContext(messages)
|
||||
context = LLMContext(messages, tools)
|
||||
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
|
||||
context,
|
||||
user_params=LLMUserAggregatorParams(
|
||||
@@ -146,6 +197,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
|
||||
|
||||
@user_aggregator.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(aggregator):
|
||||
logger.info(f"User turn idle")
|
||||
await idle_handler.handle_idle(aggregator)
|
||||
|
||||
@user_aggregator.event_handler("on_user_turn_started")
|
||||
|
||||
@@ -689,6 +689,9 @@ class LLMUserAggregator(LLMContextAggregator):
|
||||
if params.enable_user_speaking_frames:
|
||||
await self.broadcast_frame(UserStartedSpeakingFrame)
|
||||
|
||||
if self._user_idle_controller:
|
||||
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
if params.enable_interruptions and self._allow_interruptions:
|
||||
await self.push_interruption_task_frame_and_wait()
|
||||
|
||||
@@ -705,6 +708,9 @@ class LLMUserAggregator(LLMContextAggregator):
|
||||
if params.enable_user_speaking_frames:
|
||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||
|
||||
if self._user_idle_controller:
|
||||
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
|
||||
|
||||
await self._maybe_emit_user_turn_stopped(strategy)
|
||||
|
||||
async def _on_user_turn_stop_timeout(self, controller):
|
||||
|
||||
@@ -10,12 +10,14 @@ import asyncio
|
||||
from typing import Optional
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
Frame,
|
||||
FunctionCallCancelFrame,
|
||||
FunctionCallResultFrame,
|
||||
FunctionCallsStartedFrame,
|
||||
UserSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.utils.asyncio.task_manager import BaseTaskManager
|
||||
from pipecat.utils.base_object import BaseObject
|
||||
@@ -25,14 +27,14 @@ 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.
|
||||
idle (not speaking) for a configured timeout period after the bot finishes
|
||||
speaking. The timer starts when BotStoppedSpeakingFrame is received and is
|
||||
cancelled when someone starts speaking again (UserStartedSpeakingFrame or
|
||||
BotStartedSpeakingFrame).
|
||||
|
||||
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.
|
||||
The timer is suppressed while a user turn is in progress to avoid false
|
||||
triggers during interruptions (where BotStoppedSpeakingFrame arrives while
|
||||
the user is still speaking).
|
||||
|
||||
Event handlers available:
|
||||
|
||||
@@ -62,11 +64,9 @@ class UserIdleController(BaseObject):
|
||||
|
||||
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._user_turn_in_progress: bool = False
|
||||
self._function_calls_in_progress: int = 0
|
||||
self._idle_timer_task: Optional[asyncio.Task] = None
|
||||
|
||||
self._register_event_handler("on_user_turn_idle", sync=True)
|
||||
|
||||
@@ -85,19 +85,10 @@ class UserIdleController(BaseObject):
|
||||
"""
|
||||
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
|
||||
await self._cancel_idle_timer()
|
||||
|
||||
async def process_frame(self, frame: Frame):
|
||||
"""Process an incoming frame to track user activity state.
|
||||
@@ -105,69 +96,52 @@ class UserIdleController(BaseObject):
|
||||
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)
|
||||
if isinstance(frame, BotStoppedSpeakingFrame):
|
||||
# Only start the timer if the user isn't mid-turn and no function
|
||||
# calls are pending.
|
||||
#
|
||||
# Interruption case: the frame order is UserStartedSpeaking →
|
||||
# BotStoppedSpeaking → (user keeps talking) → UserStoppedSpeaking.
|
||||
# Without the user-turn guard the timer would start while the user
|
||||
# is still speaking.
|
||||
#
|
||||
# Function call case: normally FunctionCallsStarted arrives after
|
||||
# BotStoppedSpeaking and cancels the timer directly. But a race
|
||||
# condition can cause FunctionCallsStarted to arrive before
|
||||
# BotStoppedSpeaking when pushing a TTSSpeakFrame in the
|
||||
# on_function_calls_started event handler, so the counter guard
|
||||
# prevents the timer from starting while a function call is in progress.
|
||||
if not self._user_turn_in_progress and self._function_calls_in_progress == 0:
|
||||
await self._start_idle_timer()
|
||||
elif isinstance(frame, BotStartedSpeakingFrame):
|
||||
await self._cancel_idle_timer()
|
||||
elif isinstance(frame, UserStartedSpeakingFrame):
|
||||
self._user_turn_in_progress = True
|
||||
await self._cancel_idle_timer()
|
||||
elif isinstance(frame, UserStoppedSpeakingFrame):
|
||||
self._user_turn_in_progress = False
|
||||
elif isinstance(frame, FunctionCallsStartedFrame):
|
||||
await self._handle_function_calls_started(frame)
|
||||
elif isinstance(frame, FunctionCallResultFrame):
|
||||
await self._handle_function_call_result(frame)
|
||||
self._function_calls_in_progress += len(frame.function_calls)
|
||||
await self._cancel_idle_timer()
|
||||
elif isinstance(frame, (FunctionCallResultFrame, FunctionCallCancelFrame)):
|
||||
self._function_calls_in_progress = max(0, self._function_calls_in_progress - 1)
|
||||
|
||||
async def _handle_activity(self, _: UserSpeakingFrame | BotSpeakingFrame):
|
||||
"""Handle continuous activity frames that should reset the idle timer.
|
||||
async def _start_idle_timer(self):
|
||||
"""Start (or restart) the idle timer."""
|
||||
await self._cancel_idle_timer()
|
||||
self._idle_timer_task = self.task_manager.create_task(
|
||||
self._idle_timer_expired(),
|
||||
f"{self}::idle_timer",
|
||||
)
|
||||
|
||||
These frames are emitted continuously while the user or bot is speaking,
|
||||
so we simply reset the timer whenever we receive them.
|
||||
async def _cancel_idle_timer(self):
|
||||
"""Cancel the idle timer if running."""
|
||||
if self._idle_timer_task:
|
||||
await self.task_manager.cancel_task(self._idle_timer_task)
|
||||
self._idle_timer_task = None
|
||||
|
||||
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_turn_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_turn_idle")
|
||||
async def _idle_timer_expired(self):
|
||||
"""Sleep for the timeout duration then fire the idle event."""
|
||||
await asyncio.sleep(self._user_idle_timeout)
|
||||
self._idle_timer_task = None
|
||||
await self._call_event_handler("on_user_turn_idle")
|
||||
|
||||
@@ -189,6 +189,9 @@ class UserTurnProcessor(FrameProcessor):
|
||||
if params.enable_user_speaking_frames:
|
||||
await self.broadcast_frame(UserStartedSpeakingFrame)
|
||||
|
||||
if self._user_idle_controller:
|
||||
await self._user_idle_controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
if params.enable_interruptions and self._allow_interruptions:
|
||||
await self.push_interruption_task_frame_and_wait()
|
||||
|
||||
@@ -205,6 +208,9 @@ class UserTurnProcessor(FrameProcessor):
|
||||
if params.enable_user_speaking_frames:
|
||||
await self.broadcast_frame(UserStoppedSpeakingFrame)
|
||||
|
||||
if self._user_idle_controller:
|
||||
await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame())
|
||||
|
||||
await self._call_event_handler("on_user_turn_stopped", strategy)
|
||||
|
||||
async def _on_user_turn_stop_timeout(self, controller):
|
||||
|
||||
@@ -6,12 +6,13 @@
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
import unittest.mock
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
FunctionCallResultFrame,
|
||||
FunctionCallsStartedFrame,
|
||||
UserSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
)
|
||||
from pipecat.turns.user_idle_controller import UserIdleController
|
||||
@@ -25,8 +26,8 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
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."""
|
||||
async def test_idle_after_bot_stops_speaking(self):
|
||||
"""Test that idle event fires after BotStoppedSpeakingFrame + timeout."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -37,18 +38,16 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# 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."""
|
||||
async def test_user_speaking_cancels_timer(self):
|
||||
"""Test that UserStartedSpeakingFrame cancels the idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -59,20 +58,18 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
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())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_bot_speaking_resets_idle_timer(self):
|
||||
"""Test that BotSpeakingFrame frames reset the idle timer."""
|
||||
async def test_bot_speaking_cancels_timer(self):
|
||||
"""Test that BotStartedSpeakingFrame cancels the idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -83,102 +80,61 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_no_idle_before_bot_speaks(self):
|
||||
"""Test that idle does not fire if no BotStoppedSpeakingFrame is received."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Wait without any frames
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_interruption_no_false_trigger(self):
|
||||
"""Test that BotStoppedSpeakingFrame during a user turn does not start the timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# User starts speaking (interruption)
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
# Bot stops speaking due to interruption
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# 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_turn_idle")
|
||||
async def on_user_turn_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_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Wait without starting conversation
|
||||
# Wait - timer should NOT have started because user turn is in progress
|
||||
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_turn_idle")
|
||||
async def on_user_turn_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."""
|
||||
async def test_idle_cycle(self):
|
||||
"""Test that idle fires, then can fire again after another bot speaking cycle."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -189,29 +145,105 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_count
|
||||
idle_count += 1
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
# First idle
|
||||
# First cycle: bot stops → idle fires
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
first_count = idle_count
|
||||
self.assertGreaterEqual(first_count, 1)
|
||||
self.assertEqual(idle_count, 1)
|
||||
|
||||
# Second idle
|
||||
# Second cycle: bot starts → bot stops → idle fires again
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
second_count = idle_count
|
||||
self.assertGreater(second_count, first_count)
|
||||
self.assertEqual(idle_count, 2)
|
||||
|
||||
# User activity resets timer
|
||||
await controller.process_frame(UserSpeakingFrame())
|
||||
await controller.cleanup()
|
||||
|
||||
# Give a moment for the timer to reset
|
||||
await asyncio.sleep(0.1)
|
||||
async def test_cleanup_cancels_timer(self):
|
||||
"""Test that cleanup cancels a pending idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.cleanup()
|
||||
|
||||
# Third idle
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
third_count = idle_count
|
||||
self.assertGreater(third_count, second_count)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
async def test_function_call_cancels_timer(self):
|
||||
"""Test normal ordering: BotStopped starts timer, FunctionCallsStarted cancels it."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Bot finishes speaking, timer starts
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
# Function call starts shortly after, cancels the timer
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.process_frame(
|
||||
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
|
||||
)
|
||||
|
||||
# Wait longer than timeout — should not fire
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_function_call_suppresses_timer(self):
|
||||
"""Test race condition: FunctionCallsStarted arrives before BotStopped.
|
||||
|
||||
A race condition can cause FunctionCallsStarted to arrive before
|
||||
BotStoppedSpeaking. The counter guard prevents the timer from starting
|
||||
while a function call is in progress.
|
||||
"""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# LLM emits function call and "let me check" concurrently
|
||||
await controller.process_frame(
|
||||
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
|
||||
)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Wait longer than timeout — should not fire (function call in progress)
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
# Function call completes, bot speaks result
|
||||
await controller.process_frame(
|
||||
FunctionCallResultFrame(
|
||||
function_name="test", tool_call_id="123", arguments={}, result="ok"
|
||||
)
|
||||
)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Now the timer should start and fire
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user