feat: enhance escalation traces (#522)

Author: dianagrecu-uipath

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-langchain"
-version = "0.5.29"
+version = "0.5.30"
 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

@@ -21,7 +21,7 @@
 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.platform.common import CreateEscalation, UiPathConfig
 from uipath.runtime.errors import UiPathErrorCode
 
 from uipath_langchain.agent.react.jsonschema_pydantic_converter import create_model
@@ -107,6 +107,48 @@ def _get_user_email(user: Any) -> str | None:
     return getattr(user, "emailAddress", None)
 
 
+def _parse_task_data(
+    data: dict[str, Any],
+    input_schema: dict[str, Any],
+    output_schema: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """
+    Filter action center task data based on input/output schemas.
+
+    When output_schema is None, returns only fields not present in input_schema.
+    When output_schema is provided, returns only fields defined in output_schema.
+
+    Args:
+        data: Raw task data from action center
+        input_schema: JSON schema defining the input fields
+        output_schema: Optional JSON schema defining expected output fields
+
+    Returns:
+        Filtered dictionary containing only relevant output fields
+    """
+    filtered_fields: dict[str, Any] = {}
+
+    if output_schema is None:
+        input_field_names = set()
+        if "properties" in input_schema:
+            input_field_names = set(input_schema["properties"].keys())
+
+        for field_name, field_value in data.items():
+            if field_name not in input_field_names:
+                filtered_fields[field_name] = field_value
+
+    else:
+        output_field_names = set()
+        if "properties" in output_schema:
+            output_field_names = set(output_schema["properties"].keys())
+
+        for field_name, field_value in data.items():
+            if field_name in output_field_names:
+                filtered_fields[field_name] = field_value
+
+    return filtered_fields
+
+
 def create_escalation_tool(
     resource: AgentEscalationResourceConfig,
 ) -> StructuredTool:
@@ -161,25 +203,31 @@ async def escalate():
 
         # Extract task info before validation
         task_id = result.id
+        task_url = f"{UiPathConfig.base_url}/actions_/tasks/{task_id}"
         assigned_to = _get_user_email(result.assigned_to_user)
 
-        escalation_action = result.action
-        escalation_output = result.data or {}
+        outcome = result.action
+        escalation_output = _parse_task_data(
+            result.data,
+            input_schema=input_model.model_json_schema(),
+            output_schema=EscalationToolOutput.model_json_schema(),
+        )
 
         outcome_str = (
-            channel.outcome_mapping.get(escalation_action)
-            if channel.outcome_mapping and escalation_action
+            channel.outcome_mapping.get(outcome)
+            if channel.outcome_mapping and outcome
             else None
         )
-        outcome = (
+        escalation_action = (
             EscalationAction(outcome_str) if outcome_str else EscalationAction.CONTINUE
         )
 
         return {
-            "action": outcome,
+            "action": escalation_action,
             "output": escalation_output,
-            "outcome": escalation_action,
+            "outcome": outcome,
             "task_id": task_id,
+            "task_url": task_url,
             "assigned_to": assigned_to,
         }
 
@@ -215,7 +263,7 @@ async def escalation_wrapper(
             )
 
         return {
-            **result["output"],
+            "output": result["output"],
             "outcome": result["outcome"],
             "task_id": result["task_id"],
             "assigned_to": result["assigned_to"],

tests/agent/tools/test_escalation_tool.py

@@ -16,6 +16,7 @@
 
 from uipath_langchain.agent.tools.escalation_tool import (
     _get_user_email,
+    _parse_task_data,
     create_escalation_tool,
     resolve_asset,
     resolve_recipient_value,
@@ -692,3 +693,38 @@ async def test_wrapper_handles_missing_assigned_to_user(
 
         assert result["task_id"] == 99999
         assert result["assigned_to"] is None
+
+
+class TestParseTaskData:
+    """Test output task data is filtered correctly."""
+
+    def test_filters_input_fields_when_no_output_schema(self):
+        """Test that input fields are excluded when output_schema is None."""
+        data = {"input_field": "value1", "output_field": "value2"}
+        input_schema = {"properties": {"input_field": {"type": "string"}}}
+
+        result = _parse_task_data(data, input_schema, output_schema=None)
+
+        assert result == {"output_field": "value2"}
+        assert "input_field" not in result
+
+    def test_includes_only_output_fields_when_output_schema_provided(self):
+        """Test that only output schema fields are included."""
+        data = {"field1": "a", "field2": "b", "field3": "c"}
+        input_schema = {"properties": {"field1": {"type": "string"}}}
+        output_schema = {
+            "properties": {"field1": {"type": "string"}, "field2": {"type": "string"}}
+        }
+
+        result = _parse_task_data(data, input_schema, output_schema)
+
+        assert result == {"field1": "a", "field2": "b"}
+        assert "field3" not in result
+
+    def test_handles_missing_properties_in_schemas(self):
+        """Test behavior when schemas lack 'properties' key."""
+        data = {"field": "value"}
+
+        # No properties key in schemas
+        result = _parse_task_data(data, {}, None)
+        assert result == {"field": "value"}