fix(voice): bound EndCallTool wait for tool-reply playout

livekit-agents/livekit/agents/beta/tools/end_call.py

@@ -23,6 +23,12 @@
 Don't generate any other text or response when the tool is called.
 """
 
+# Bound both the wait for the tool-reply speech_created event and the wait for that
+# speech to finish. Guarding only the first await left a hang path: speech_created
+# fires, but await speech_handle never returns, finally never runs, and shutdown is
+# skipped (room/SIP stay up when delete_room=True). See #5096.
+TOOL_REPLY_TIMEOUT = 5.0
+
 
 class EndCallTool(Toolset):
     def __init__(
@@ -99,20 +105,27 @@ def _on_speech_done(_: SpeechHandle) -> None:
     async def _delayed_session_shutdown(self, ctx: RunContext) -> None:
         """Shutdown the session after the tool reply is played out"""
         speech_created_fut = asyncio.Future[SpeechHandle]()
+        speech_handle: SpeechHandle | None = None
 
         @ctx.session.once("speech_created")
         def _on_speech_created(ev: SpeechCreatedEvent) -> None:
             if not speech_created_fut.done():
                 speech_created_fut.set_result(ev.speech_handle)
 
         try:
-            speech_handle = await asyncio.wait_for(speech_created_fut, timeout=5.0)
-            await speech_handle
+            speech_handle = await asyncio.wait_for(speech_created_fut, timeout=TOOL_REPLY_TIMEOUT)
+            await asyncio.wait_for(speech_handle, timeout=TOOL_REPLY_TIMEOUT)
         except asyncio.TimeoutError:
             logger.warning("tool reply timed out, shutting down session")
+            # Default shutdown drains and can wait on the same unfinished reply.
+            # Force-interrupt and skip drain so room/SIP cleanup still runs (#5096).
+            if speech_handle is not None and not speech_handle.done():
+                speech_handle.interrupt(force=True)
+            ctx.session.shutdown(drain=False)
+        else:
+            ctx.session.shutdown()
         finally:
             ctx.session.off("speech_created", _on_speech_created)
-            ctx.session.shutdown()
 
     def _on_session_close(self, ev: CloseEvent) -> None:
         """Close the job process when AgentSession is closed"""

tests/test_tools.py

@@ -336,6 +336,125 @@ def test_end_call_tool_ignore_on_enter_flag():
     assert tool.info.flags & ToolFlag.IGNORE_ON_ENTER
 
 
+class _FakeEndCallSession:
+    """Minimal session surface for EndCallTool._delayed_session_shutdown."""
+
+    def __init__(self) -> None:
+        self.shutdown_calls: list[bool] = []
+        self._handlers: dict[str, list[Any]] = {}
+
+    def once(self, event: str, callback: Any = None) -> Any:
+        if callback is None:
+
+            def decorator(cb: Any) -> Any:
+                self._handlers.setdefault(event, []).append(cb)
+                return cb
+
+            return decorator
+
+        self._handlers.setdefault(event, []).append(callback)
+        return callback
+
+    def off(self, event: str, callback: Any) -> None:
+        handlers = self._handlers.get(event)
+        if handlers and callback in handlers:
+            handlers.remove(callback)
+
+    def emit(self, event: str, ev: Any) -> None:
+        for callback in list(self._handlers.get(event, [])):
+            callback(ev)
+
+    def shutdown(self, *, drain: bool = True) -> None:
+        self.shutdown_calls.append(drain)
+
+
+@pytest.mark.asyncio
+async def test_delayed_session_shutdown_times_out_when_speech_handle_hangs() -> None:
+    """speech_created fired but playout never finishes must still shut down (#5096)."""
+    import asyncio
+    from types import SimpleNamespace
+
+    from livekit.agents.beta import EndCallTool
+    from livekit.agents.beta.tools import end_call as end_call_mod
+    from livekit.agents.voice.events import SpeechCreatedEvent
+    from livekit.agents.voice.speech_handle import SpeechHandle
+
+    tool = EndCallTool(delete_room=False)
+    session = _FakeEndCallSession()
+    ctx = SimpleNamespace(session=session)
+
+    task = asyncio.create_task(tool._delayed_session_shutdown(ctx))
+    await asyncio.sleep(0)
+
+    hanging = SpeechHandle.create()
+    session.emit(
+        "speech_created",
+        SpeechCreatedEvent(
+            user_initiated=False,
+            source="generate_reply",
+            speech_handle=hanging,
+        ),
+    )
+
+    # Without a timeout on `await speech_handle`, this never returns — even under
+    # virtual time, because no timer is scheduled for a bare Future.
+    await asyncio.wait_for(task, timeout=end_call_mod.TOOL_REPLY_TIMEOUT + 1.0)
+    assert session.shutdown_calls == [False]
+    assert hanging.interrupted
+
+
+@pytest.mark.asyncio
+async def test_delayed_session_shutdown_times_out_when_speech_created_never_fires() -> None:
+    import asyncio
+    from types import SimpleNamespace
+
+    from livekit.agents.beta import EndCallTool
+    from livekit.agents.beta.tools import end_call as end_call_mod
+
+    tool = EndCallTool(delete_room=False)
+    session = _FakeEndCallSession()
+    ctx = SimpleNamespace(session=session)
+
+    await asyncio.wait_for(
+        tool._delayed_session_shutdown(ctx),
+        timeout=end_call_mod.TOOL_REPLY_TIMEOUT + 1.0,
+    )
+    assert session.shutdown_calls == [False]
+
+
+@pytest.mark.asyncio
+async def test_delayed_session_shutdown_waits_for_completed_speech_handle() -> None:
+    import asyncio
+    from types import SimpleNamespace
+
+    from livekit.agents.beta import EndCallTool
+    from livekit.agents.voice.events import SpeechCreatedEvent
+    from livekit.agents.voice.speech_handle import SpeechHandle
+
+    tool = EndCallTool(delete_room=False)
+    session = _FakeEndCallSession()
+    ctx = SimpleNamespace(session=session)
+
+    task = asyncio.create_task(tool._delayed_session_shutdown(ctx))
+    await asyncio.sleep(0)
+
+    handle = SpeechHandle.create()
+    session.emit(
+        "speech_created",
+        SpeechCreatedEvent(
+            user_initiated=False,
+            source="generate_reply",
+            speech_handle=handle,
+        ),
+    )
+    await asyncio.sleep(0)
+    assert session.shutdown_calls == []
+
+    handle._mark_done()
+    await asyncio.wait_for(task, timeout=1.0)
+    assert session.shutdown_calls == [True]
+
+
 class TestToolExecution:
     def test_function_arguments_to_pydantic_model(self):
         schema1 = function_arguments_to_pydantic_model(mock_tool_1)