feat: escalations parity

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-langchain"
-version = "0.5.13"
+version = "0.5.14"
 description = "Python SDK that enables developers to build and deploy LangGraph agents to the UiPath Cloud Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath_langchain/agent/tools/escalation_tool.py

@@ -13,15 +13,17 @@
     AgentEscalationResourceConfig,
     AssetRecipient,
     StandardRecipient,
+    TaskTitle,
+    TextBuilderTaskTitle,
 )
+from uipath.agent.utils.text_tokens import build_string_from_tokens
 from uipath.eval.mocks import mockable
 from uipath.platform import UiPath
 from uipath.platform.action_center.tasks import TaskRecipient, TaskRecipientType
 from uipath.platform.common import CreateEscalation
 from uipath.runtime.errors import UiPathErrorCode
 
 from uipath_langchain.agent.react.jsonschema_pydantic_converter import create_model
-from uipath_langchain.agent.react.types import AgentGraphState
 from uipath_langchain.agent.tools.static_args import (
     handle_static_args,
 )
@@ -30,8 +32,9 @@
 )
 
 from ..exceptions import AgentTerminationException
+from ..react.types import AgentGraphState
 from .tool_node import ToolWrapperReturnType
-from .utils import sanitize_tool_name
+from .utils import sanitize_dict_for_serialization, sanitize_tool_name
 
 
 class EscalationAction(str, Enum):
@@ -81,6 +84,19 @@ async def resolve_asset(asset_name: str, folder_path: str) -> str | None:
         ) from e
 
 
+def _resolve_task_title(
+    task_title: TaskTitle | str | None, agent_input: dict[str, Any]
+) -> str:
+    """Resolve task title based on channel configuration."""
+    if isinstance(task_title, TextBuilderTaskTitle):
+        return build_string_from_tokens(task_title.tokens, agent_input)
+
+    if isinstance(task_title, str):
+        return task_title
+
+    return "Escalation Task"
+
+
 def create_escalation_tool(
     resource: AgentEscalationResourceConfig,
 ) -> StructuredTool:
@@ -100,17 +116,17 @@ def create_escalation_tool(
         example_calls=channel.properties.example_calls,
     )
     async def escalation_tool_fn(**kwargs: Any) -> dict[str, Any]:
-        task_title = channel.task_title or "Escalation Task"
-
         recipient: TaskRecipient | None = (
             await resolve_recipient_value(channel.recipients[0])
             if channel.recipients
             else None
         )
 
-        # Recipient requires runtime resolution, store in metadata after resolving
+        task_title = "Escalation Task"
         if tool.metadata is not None:
+            # Recipient requires runtime resolution, store in metadata after resolving
             tool.metadata["recipient"] = recipient
+            task_title = tool.metadata.get("task_title") or task_title
 
         result = interrupt(
             CreateEscalation(
@@ -119,7 +135,6 @@ async def escalation_tool_fn(**kwargs: Any) -> dict[str, Any]:
                 recipient=recipient,
                 app_name=channel.properties.app_name,
                 app_folder_path=channel.properties.folder_name,
-                app_version=channel.properties.app_version,
                 priority=channel.priority,
                 labels=channel.labels,
                 is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
@@ -150,6 +165,13 @@ async def escalation_wrapper(
         call: ToolCall,
         state: AgentGraphState,
     ) -> ToolWrapperReturnType:
+        if tool.metadata is None:
+            raise RuntimeError("Tool metadata is required for task_title resolution")
+
+        tool.metadata["task_title"] = _resolve_task_title(
+            channel.task_title, sanitize_dict_for_serialization(dict(state))
+        )
+
         call["args"] = handle_static_args(resource, state, call["args"])
         result = await tool.ainvoke(call["args"])
 
@@ -179,7 +201,7 @@ async def escalation_wrapper(
             "tool_type": "escalation",
             "display_name": channel.properties.app_name,
             "channel_type": channel.type,
-            "assignee": None,
+            "recipient": None,
         },
     )
     tool.set_tool_wrappers(awrapper=escalation_wrapper)

tests/agent/tools/test_escalation_tool.py

@@ -3,6 +3,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from langchain_core.messages import ToolCall
 from uipath.agent.models.agent import (
     AgentEscalationChannel,
     AgentEscalationChannelProperties,
@@ -281,8 +282,11 @@ async def test_escalation_tool_metadata_has_recipient(
 
         tool = create_escalation_tool(escalation_resource)
 
-        # Invoke the tool to trigger assignee resolution
-        await tool.ainvoke({})
+        # Create mock state and call to invoke through wrapper
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        # Invoke through the wrapper to test full flow
+        await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
 
         assert tool.metadata is not None
         assert tool.metadata["recipient"] == TaskRecipient(
@@ -303,8 +307,148 @@ async def test_escalation_tool_metadata_recipient_none_when_no_recipients(
 
         tool = create_escalation_tool(escalation_resource_no_recipient)
 
-        # Invoke the tool to trigger assignee resolution
-        await tool.ainvoke({})
+        # Create mock state and call to invoke through wrapper
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        # Invoke through the wrapper to test full flow
+        await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
 
         assert tool.metadata is not None
         assert tool.metadata["recipient"] is None
+
+    @pytest.mark.asyncio
+    @patch("uipath_langchain.agent.tools.escalation_tool.interrupt")
+    async def test_escalation_tool_with_string_task_title(self, mock_interrupt):
+        """Test escalation tool with legacy string task title."""
+        mock_result = MagicMock()
+        mock_result.action = None
+        mock_result.data = {}
+        mock_interrupt.return_value = mock_result
+
+        # Create resource with string task title
+        channel_dict = {
+            "name": "action_center",
+            "type": "actionCenter",
+            "description": "Action Center channel",
+            "inputSchema": {"type": "object", "properties": {}},
+            "outputSchema": {"type": "object", "properties": {}},
+            "properties": {
+                "appName": "ApprovalApp",
+                "appVersion": 1,
+                "resourceKey": "test-key",
+            },
+            "recipients": [],
+            "taskTitle": "Static Task Title",
+        }
+
+        resource = AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[AgentEscalationChannel(**channel_dict)],
+        )
+
+        tool = create_escalation_tool(resource)
+
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        # Invoke through the wrapper to test full flow
+        await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
+
+        # Verify interrupt was called with the static title
+        call_args = mock_interrupt.call_args[0][0]
+        assert call_args.title == "Static Task Title"
+
+    @pytest.mark.asyncio
+    @patch("uipath_langchain.agent.tools.escalation_tool.interrupt")
+    async def test_escalation_tool_with_text_builder_task_title(self, mock_interrupt):
+        """Test escalation tool with TEXT_BUILDER task title builds from tokens."""
+        mock_result = MagicMock()
+        mock_result.action = None
+        mock_result.data = {}
+        mock_interrupt.return_value = mock_result
+
+        # Create resource with TEXT_BUILDER task title containing variable token
+        channel_dict = {
+            "name": "action_center",
+            "type": "actionCenter",
+            "description": "Action Center channel",
+            "inputSchema": {"type": "object", "properties": {}},
+            "outputSchema": {"type": "object", "properties": {}},
+            "properties": {
+                "appName": "ApprovalApp",
+                "appVersion": 1,
+                "resourceKey": "test-key",
+            },
+            "recipients": [],
+            "taskTitle": {
+                "type": "textBuilder",
+                "tokens": [
+                    {"type": "simpleText", "rawString": "Approve request for "},
+                    {"type": "variable", "rawString": "input.userName"},
+                ],
+            },
+        }
+
+        resource = AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[AgentEscalationChannel(**channel_dict)],
+        )
+
+        tool = create_escalation_tool(resource)
+
+        # Create mock state with variables for token interpolation
+        state = {"userName": "John Doe", "messages": []}
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        # Invoke through the wrapper to test full flow
+        await tool.awrapper(tool, call, state)  # type: ignore[attr-defined]
+
+        # Verify interrupt was called with the correctly built task title
+        assert mock_interrupt.called
+        call_args = mock_interrupt.call_args[0][0]
+        assert call_args.title == "Approve request for John Doe"
+
+    @pytest.mark.asyncio
+    @patch("uipath_langchain.agent.tools.escalation_tool.interrupt")
+    async def test_escalation_tool_with_empty_task_title_defaults_to_escalation_task(
+        self, mock_interrupt
+    ):
+        """Test escalation tool defaults to 'Escalation Task' when task title is empty."""
+        mock_result = MagicMock()
+        mock_result.action = None
+        mock_result.data = {}
+        mock_interrupt.return_value = mock_result
+
+        # Create resource with empty string task title
+        channel_dict = {
+            "name": "action_center",
+            "type": "actionCenter",
+            "description": "Action Center channel",
+            "inputSchema": {"type": "object", "properties": {}},
+            "outputSchema": {"type": "object", "properties": {}},
+            "properties": {
+                "appName": "ApprovalApp",
+                "appVersion": 1,
+                "resourceKey": "test-key",
+            },
+            "recipients": [],
+            "taskTitle": "",
+        }
+
+        resource = AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[AgentEscalationChannel(**channel_dict)],
+        )
+
+        tool = create_escalation_tool(resource)
+
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        # Invoke through the wrapper to test full flow
+        await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
+
+        # Verify interrupt was called with the default title
+        call_args = mock_interrupt.call_args[0][0]
+        assert call_args.title == "Escalation Task"