feat(workflows): introduce structured WarmTransferError with SIP facts (#7200)

livekit-agents/livekit/agents/beta/workflows/__init__.py

@@ -7,7 +7,13 @@
 from .phone_number import GetPhoneNumberResult, GetPhoneNumberTask
 from .task_group import TaskCompletedEvent, TaskGroup, TaskGroupResult
 from .utils import WorkflowInstructions
-from .warm_transfer import TwilioConnectorWarmTransferTask, WarmTransferResult, WarmTransferTask
+from .warm_transfer import (
+    TwilioConnectorWarmTransferTask,
+    WarmTransferError,
+    WarmTransferFailure,
+    WarmTransferResult,
+    WarmTransferTask,
+)
 
 __all__ = [
     "GetEmailTask",
@@ -31,4 +37,6 @@
     "WarmTransferTask",
     "TwilioConnectorWarmTransferTask",
     "WarmTransferResult",
+    "WarmTransferError",
+    "WarmTransferFailure",
 ]

livekit-agents/livekit/agents/beta/workflows/warm_transfer.py

@@ -4,7 +4,8 @@
 import contextlib
 import os
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from enum import Enum
+from typing import TYPE_CHECKING, Any
 from xml.sax.saxutils import quoteattr
 
 from livekit import api, rtc
@@ -32,6 +33,32 @@
     from ...voice.turn import TurnDetectionMode
 
 
+class WarmTransferFailure(str, Enum):
+    DIAL_FAILED = "dial_failed"
+    DESTINATION_LEFT = "destination_left"
+    DECLINED = "declined"
+    VOICEMAIL = "voicemail"
+    ROOM_CLOSED = "room_closed"
+    CALLER_LEFT = "caller_left"
+
+
+class WarmTransferError(ToolError):
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: WarmTransferFailure,
+        disconnect_reason: rtc.DisconnectReason.ValueType | None = None,
+        call_status: str | None = None,
+        reason: str | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.disconnect_reason = disconnect_reason
+        self.call_status = call_status
+        self.reason = reason
+
+
 @dataclass
 class WarmTransferResult:
     human_agent_identity: str
@@ -118,6 +145,11 @@ def __init__(
         self._human_agent_sess: AgentSession | None = None
         self._human_agent_failed_fut: asyncio.Future[None] = asyncio.Future()
         self._human_agent_identity = "human-agent-sip"
+        self._destination_disconnect_reason: rtc.DisconnectReason.ValueType | None = None
+        self._destination_call_status: str | None = None
+        self._human_agent_participant_disconnected_cb: Any | None = (
+            self._on_human_agent_participant_disconnected
+        )
 
         self._setup_origination(
             sip_call_to=sip_call_to,
@@ -220,7 +252,10 @@ async def on_enter(self) -> None:
 
         except Exception:
             logger.exception("could not dial human agent")
-            self._set_result(ToolError("could not dial human agent"))
+            err = WarmTransferError(
+                "could not dial human agent", code=WarmTransferFailure.DIAL_FAILED
+            )
+            self._set_result(err)
             return
 
         finally:
@@ -245,12 +280,23 @@ async def decline_transfer(self, reason: str) -> None:
         Args:
             reason: A short explanation of why the human agent declined to connect to the caller
         """
-        self._set_result(ToolError(f"human agent declined to connect: {reason}"))
+        self._set_result(
+            WarmTransferError(
+                f"human agent declined to connect: {reason}",
+                code=WarmTransferFailure.DECLINED,
+                reason=reason,
+            )
+        )
 
     @function_tool(flags=ToolFlag.IGNORE_ON_ENTER)
     async def voicemail_detected(self) -> None:
         """Called when the call reaches voicemail. Use this tool AFTER you hear the voicemail greeting"""
-        self._set_result(ToolError("voicemail detected"))
+        self._set_result(
+            WarmTransferError(
+                "voicemail detected",
+                code=WarmTransferFailure.VOICEMAIL,
+            )
+        )
 
     def _on_human_agent_room_close(self, reason: rtc.DisconnectReason.ValueType) -> None:
         logger.debug(
@@ -260,7 +306,44 @@ def _on_human_agent_room_close(self, reason: rtc.DisconnectReason.ValueType) ->
         with contextlib.suppress(asyncio.InvalidStateError):
             self._human_agent_failed_fut.set_result(None)
 
-        self._set_result(ToolError(f"room closed: {rtc.DisconnectReason.Name(reason)}"))
+        if self._destination_disconnect_reason is not None:
+            self._set_result(
+                WarmTransferError(
+                    f"destination left: {rtc.DisconnectReason.Name(self._destination_disconnect_reason)}",
+                    code=WarmTransferFailure.DESTINATION_LEFT,
+                    disconnect_reason=self._destination_disconnect_reason,
+                    call_status=self._destination_call_status,
+                )
+            )
+        else:
+            self._set_result(
+                WarmTransferError(
+                    f"room closed: {rtc.DisconnectReason.Name(reason)}",
+                    code=WarmTransferFailure.ROOM_CLOSED,
+                    disconnect_reason=reason,
+                    call_status=self._destination_call_status,
+                )
+            )
+
+    def _on_human_agent_participant_disconnected(self, participant: rtc.RemoteParticipant) -> None:
+        if participant.identity == self._human_agent_identity:
+            self._destination_disconnect_reason = participant.disconnect_reason
+            self._destination_call_status = participant.attributes.get("sip.callStatus")
+            with contextlib.suppress(asyncio.InvalidStateError):
+                self._human_agent_failed_fut.set_result(None)
+            reason_name = (
+                rtc.DisconnectReason.Name(participant.disconnect_reason)
+                if participant.disconnect_reason is not None
+                else "UNKNOWN_REASON"
+            )
+            self._set_result(
+                WarmTransferError(
+                    f"destination left: {reason_name}",
+                    code=WarmTransferFailure.DESTINATION_LEFT,
+                    disconnect_reason=participant.disconnect_reason,
+                    call_status=self._destination_call_status,
+                )
+            )
 
     def _on_caller_participant_disconnected(self, participant: rtc.RemoteParticipant) -> None:
         if participant.kind not in DEFAULT_PARTICIPANT_KINDS:
@@ -320,6 +403,7 @@ async def _dial_human_agent(self) -> AgentSession:
 
         # if human agent hung up for whatever reason, we'd resume the caller conversation
         room.on("disconnected", self._on_human_agent_room_close)
+        room.on("participant_disconnected", self._on_human_agent_participant_disconnected)
 
         human_agent_sess: AgentSession = AgentSession(
             vad=self.session.vad or NOT_GIVEN,
@@ -358,7 +442,7 @@ async def _dial_human_agent(self) -> AgentSession:
                 identity=self._human_agent_identity,
                 room=room,
             )
-        except Exception:
+        except BaseException:
             human_agent_sess.shutdown()
             raise
 
@@ -392,15 +476,27 @@ async def _merge_calls(self) -> None:
         human_agent_room = self._human_agent_sess.room_io.room
         # we no longer care about the human agent session. it's supposed to be over
         human_agent_room.off("disconnected", self._on_human_agent_room_close)
+        if self._human_agent_participant_disconnected_cb is not None:
+            human_agent_room.off(
+                "participant_disconnected", self._human_agent_participant_disconnected_cb
+            )
 
         logger.debug(f"moving {self._human_agent_identity} to caller room {self._caller_room.name}")
-        await job_ctx.api.room.move_participant(
-            api.MoveParticipantRequest(
-                room=human_agent_room.name,
-                identity=self._human_agent_identity,
-                destination_room=self._caller_room.name,
+        try:
+            await job_ctx.api.room.move_participant(
+                api.MoveParticipantRequest(
+                    room=human_agent_room.name,
+                    identity=self._human_agent_identity,
+                    destination_room=self._caller_room.name,
+                )
             )
-        )
+        except Exception:
+            human_agent_room.on("disconnected", self._on_human_agent_room_close)
+            if self._human_agent_participant_disconnected_cb is not None:
+                human_agent_room.on(
+                    "participant_disconnected", self._human_agent_participant_disconnected_cb
+                )
+            raise
 
     def _set_io_enabled(self, enabled: bool) -> None:
         input = self.session.input

tests/test_warm_transfer_error.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+
+from livekit import rtc
+from livekit.agents.beta.workflows.warm_transfer import (
+    WarmTransferError,
+    WarmTransferFailure,
+    WarmTransferTask,
+)
+from livekit.agents.llm.tool_context import ToolError
+
+pytestmark = pytest.mark.unit
+
+
+def test_warm_transfer_error_inheritance_and_properties() -> None:
+    err = WarmTransferError(
+        "human agent declined to connect: busy",
+        code=WarmTransferFailure.DECLINED,
+        reason="busy",
+    )
+    assert isinstance(err, ToolError)
+    assert err.code == WarmTransferFailure.DECLINED
+    assert err.reason == "busy"
+    assert str(err) == "human agent declined to connect: busy"
+
+
+@pytest.mark.asyncio
+async def test_decline_transfer_creates_structured_error() -> None:
+    task = object.__new__(WarmTransferTask)
+    task._human_agent_sess = None
+    task._hold_audio_handle = None
+    task._set_io_enabled = MagicMock()
+    task.done = MagicMock(return_value=False)
+    task.complete = MagicMock()
+
+    await task.decline_transfer("agent is in a meeting")
+
+    task.complete.assert_called_once()
+    result = task.complete.call_args[0][0]
+    assert isinstance(result, WarmTransferError)
+    assert result.code == WarmTransferFailure.DECLINED
+    assert result.reason == "agent is in a meeting"
+    assert "human agent declined to connect: agent is in a meeting" in str(result)
+
+
+@pytest.mark.asyncio
+async def test_voicemail_detected_creates_structured_error() -> None:
+    task = object.__new__(WarmTransferTask)
+    task._human_agent_sess = None
+    task._hold_audio_handle = None
+    task._set_io_enabled = MagicMock()
+    task.done = MagicMock(return_value=False)
+    task.complete = MagicMock()
+
+    await task.voicemail_detected()
+
+    task.complete.assert_called_once()
+    result = task.complete.call_args[0][0]
+    assert isinstance(result, WarmTransferError)
+    assert result.code == WarmTransferFailure.VOICEMAIL
+    assert str(result) == "voicemail detected"
+
+
+@pytest.mark.asyncio
+async def test_human_agent_room_close_with_destination_left() -> None:
+    task = object.__new__(WarmTransferTask)
+    task._human_agent_sess = None
+    task._hold_audio_handle = None
+    task._set_io_enabled = MagicMock()
+    task.done = MagicMock(return_value=False)
+    task.complete = MagicMock()
+    task._human_agent_failed_fut = asyncio.get_running_loop().create_future()
+
+    # Pre-recorded destination departure
+    task._destination_disconnect_reason = rtc.DisconnectReason.USER_UNAVAILABLE
+    task._destination_call_status = "busy"
+
+    task._on_human_agent_room_close(rtc.DisconnectReason.ROOM_DELETED)
+
+    task.complete.assert_called_once()
+    result = task.complete.call_args[0][0]
+    assert isinstance(result, WarmTransferError)
+    assert result.code == WarmTransferFailure.DESTINATION_LEFT
+    assert result.disconnect_reason == rtc.DisconnectReason.USER_UNAVAILABLE
+    assert result.call_status == "busy"
+    assert "destination left: USER_UNAVAILABLE" in str(result)
+
+
+@pytest.mark.asyncio
+async def test_human_agent_room_close_without_destination_left() -> None:
+    task = object.__new__(WarmTransferTask)
+    task._human_agent_sess = None
+    task._hold_audio_handle = None
+    task._set_io_enabled = MagicMock()
+    task.done = MagicMock(return_value=False)
+    task.complete = MagicMock()
+    task._human_agent_failed_fut = asyncio.get_running_loop().create_future()
+
+    task._destination_disconnect_reason = None
+    task._destination_call_status = None
+
+    task._on_human_agent_room_close(rtc.DisconnectReason.SERVER_SHUTDOWN)
+
+    task.complete.assert_called_once()
+    result = task.complete.call_args[0][0]
+    assert isinstance(result, WarmTransferError)
+    assert result.code == WarmTransferFailure.ROOM_CLOSED
+    assert result.disconnect_reason == rtc.DisconnectReason.SERVER_SHUTDOWN
+    assert "room closed: SERVER_SHUTDOWN" in str(result)
+
+
+@pytest.mark.asyncio
+async def test_human_agent_participant_disconnected_completes_transfer() -> None:
+    task = object.__new__(WarmTransferTask)
+    task._human_agent_sess = None
+    task._hold_audio_handle = None
+    task._set_io_enabled = MagicMock()
+    task.done = MagicMock(return_value=False)
+    task.complete = MagicMock()
+    task._human_agent_failed_fut = asyncio.get_running_loop().create_future()
+    task._human_agent_identity = "human-agent-sip"
+
+    # Other participant disconnecting should be ignored
+    other_participant = MagicMock(spec=rtc.RemoteParticipant)
+    other_participant.identity = "other-participant"
+    task._on_human_agent_participant_disconnected(other_participant)
+    task.complete.assert_not_called()
+    assert not task._human_agent_failed_fut.done()
+
+    # Destination participant disconnecting with USER_UNAVAILABLE completes transfer
+    dest_participant = MagicMock(spec=rtc.RemoteParticipant)
+    dest_participant.identity = "human-agent-sip"
+    dest_participant.disconnect_reason = rtc.DisconnectReason.USER_UNAVAILABLE
+    dest_participant.attributes = {"sip.callStatus": "busy"}
+
+    task._on_human_agent_participant_disconnected(dest_participant)
+
+    task.complete.assert_called_once()
+    assert task._human_agent_failed_fut.done()
+    result = task.complete.call_args[0][0]
+    assert isinstance(result, WarmTransferError)
+    assert result.code == WarmTransferFailure.DESTINATION_LEFT
+    assert result.disconnect_reason == rtc.DisconnectReason.USER_UNAVAILABLE
+    assert result.call_status == "busy"
+    assert "destination left: USER_UNAVAILABLE" in str(result)
+
+
+@pytest.mark.asyncio
+async def test_twilio_connector_warm_transfer_initializes_destination_state() -> None:
+    from livekit.agents.beta.workflows.warm_transfer import TwilioConnectorWarmTransferTask
+
+    task = TwilioConnectorWarmTransferTask(
+        phone_number="+1234567890",
+        twilio_from_number="+1098765432",
+        twilio_account_sid="AC123",
+        twilio_auth_token="secret",
+    )
+    try:
+        assert hasattr(task, "_destination_disconnect_reason")
+        assert task._destination_disconnect_reason is None
+        assert hasattr(task, "_destination_call_status")
+        assert task._destination_call_status is None
+        assert hasattr(task, "_human_agent_participant_disconnected_cb")
+        assert task._human_agent_participant_disconnected_cb is not None
+        assert callable(task._human_agent_participant_disconnected_cb)
+    finally:
+        await task._background_audio._audio_mixer.aclose()