wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make UserIdleController always-on with dynamic timeout updates

Browse Source 
Always create UserIdleController (timeout=0 means disabled), removing
all Optional guards. Add UserIdleTimeoutUpdateFrame to allow changing
the idle timeout at runtime.
This commit is contained in:
Mark Backman
2026-02-14 09:47:59 -05:00
parent cb7023681f
commit 507765625f
8 changed files with 129 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
tests/test_user_idle_controller.py
View File

@@ -13,6 +13,7 @@ from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
FunctionCallResultFrame,
FunctionCallsStartedFrame,
UserIdleTimeoutUpdateFrame,
UserStartedSpeakingFrame,
)
from pipecat.turns.user_idle_controller import UserIdleController
@@ -247,6 +248,76 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
await controller.cleanup()
async def test_disabled_by_default(self):
"""Test that timeout=0 means idle detection is disabled."""
controller = UserIdleController()
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.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
async def test_enable_via_frame(self):
"""Test enabling idle detection at runtime via UserIdleTimeoutUpdateFrame."""
controller = UserIdleController()
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
# Initially disabled — no idle fires
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
# Enable idle detection
await controller.process_frame(UserIdleTimeoutUpdateFrame(timeout=USER_IDLE_TIMEOUT))
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertTrue(idle_triggered)
await controller.cleanup()
async def test_disable_via_frame(self):
"""Test disabling idle detection at runtime via UserIdleTimeoutUpdateFrame."""
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 the timer
await controller.process_frame(BotStoppedSpeakingFrame())
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
# Disable — should cancel running timer
await controller.process_frame(UserIdleTimeoutUpdateFrame(timeout=0))
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
self.assertFalse(idle_triggered)
await controller.cleanup()
if __name__ == "__main__":
unittest.main()