feat(guardrails): create new node to create hitl task in order to capture the task url (#544)

Co-authored-by: Valentina Bojan <valentina.bojan@uipath.com>

Author: valentinabojan

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-langchain"
-version = "0.5.40"
+version = "0.5.41"
 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/guardrails/actions/base_action.py

@@ -1,12 +1,18 @@
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, Sequence, Union
 
 from uipath.platform.guardrails import BaseGuardrail, GuardrailScope
 
 from uipath_langchain.agent.guardrails.types import ExecutionStage
 
 GuardrailActionNode = tuple[str, Any]
 
+# A single node or an ordered sequence of nodes that will be chained with edges.
+# When a sequence is returned the first node is the entry-point (the one the
+# guardrail evaluation node routes to on failure) and the **last** node gets the
+# outgoing edge to `next_node`.
+GuardrailActionNodes = Union[GuardrailActionNode, Sequence[GuardrailActionNode]]
+
 
 class GuardrailAction(ABC):
     """Extensible action interface producing a node to enforce the action on guardrail validation failure."""
@@ -25,6 +31,11 @@ def action_node(
         scope: GuardrailScope,
         execution_stage: ExecutionStage,
         guarded_component_name: str,
-    ) -> GuardrailActionNode:
-        """Create and return the Action node to execute on validation failure."""
+    ) -> GuardrailActionNodes:
+        """Create and return the Action node(s) to execute on validation failure.
+
+        May return a single ``(name, callable)`` tuple or an ordered sequence of
+        such tuples.  When a sequence is returned the nodes will be chained
+        together with edges in the subgraph (first → second → … → next_node).
+        """
         ...

src/uipath_langchain/agent/guardrails/actions/block_action.py

@@ -39,7 +39,7 @@ async def _node(_state: AgentGuardrailsGraphState):
             raise AgentTerminationException(
                 code=UiPathErrorCode.EXECUTION_ERROR,
                 title="Guardrail violation",
-                detail=self.reason,
+                detail=f"Execution was blocked by guardrail [{guardrail.name}], with reason: {self.reason}",
                 category=UiPathErrorCategory.USER,
             )
 

src/uipath_langchain/agent/guardrails/actions/escalate_action.py

@@ -18,7 +18,9 @@
     AssetRecipient,
     StandardRecipient,
 )
-from uipath.platform.common import CreateEscalation, UiPathConfig
+from uipath.platform import UiPath
+from uipath.platform.action_center.tasks import Task, TaskRecipient
+from uipath.platform.common import UiPathConfig, WaitEscalation
 from uipath.platform.guardrails import (
     BaseGuardrail,
     GuardrailScope,
@@ -31,7 +33,7 @@
 from ...react.utils import extract_current_tool_call_index, find_latest_ai_message
 from ..types import ExecutionStage
 from ..utils import _extract_tool_args_from_message, get_message_content
-from .base_action import GuardrailAction, GuardrailActionNode
+from .base_action import GuardrailAction, GuardrailActionNodes
 
 
 class EscalateAction(GuardrailAction):
@@ -73,19 +75,22 @@ def action_node(
         scope: GuardrailScope,
         execution_stage: ExecutionStage,
         guarded_component_name: str,
-    ) -> GuardrailActionNode:
-        """Create a HITL escalation node for the guardrail.
-
-        Args:
-            guardrail: The guardrail that triggered this escalation action.
-            scope: The guardrail scope (LLM/AGENT/TOOL).
-            execution_stage: The execution stage (PRE_EXECUTION or POST_EXECUTION).
-
-        Returns:
-            A tuple of (node_name, node_function) where the node function triggers
-            a HITL interruption and processes the escalation response.
+    ) -> GuardrailActionNodes:
+        """Create two HITL escalation nodes for the guardrail.
+
+        Returns an ordered pair of nodes that the subgraph builder will chain:
+
+        1. **create-task node** – creates the Action Center task and persists
+           the task id in ``inner_state.hitl_task_info`` so it survives across
+           graph super-steps.  Because this node completes normally its state
+           update *is* committed to the checkpoint.
+        2. **interrupt node** – calls ``interrupt(WaitEscalation(…))`` to pause
+           execution. On resume, it reads the stored task data from state
+           (skipping re-creation) and processes the escalation response.
         """
-        node_name = _get_node_name(execution_stage, guardrail, scope)
+        base_name = _get_node_name(execution_stage, guardrail, scope)
+        create_task_node_name = f"{base_name}_create_task"
+        interrupt_node_name = base_name
 
         metadata: Dict[str, Any] = {
             "guardrail": guardrail,
@@ -98,10 +103,18 @@ def action_node(
             },
         }
 
-        async def _node(
+        # Node 1: create the escalation task
+        async def _create_task_node(
             state: AgentGuardrailsGraphState,
-        ) -> Dict[str, Any] | Command[Any]:
-            # Import here to avoid circular dependency
+        ) -> Dict[str, Any]:
+            # If the task was already created (e.g. subgraph replayed this
+            # node on resume), skip creation and return as-is.
+            hitl_info = state.inner_state.hitl_task_info
+            existing_task = hitl_info.get(create_task_node_name) if hitl_info else None
+            if existing_task is not None:
+                return {}
+
+            # Lazy import to avoid circular dependency with escalation_tool
             from ...tools.escalation_tool import resolve_recipient_value
 
             # Resolve recipient value (handles both StandardRecipient and AssetRecipient)
@@ -179,12 +192,60 @@ async def _node(
                 data["ToolInputs"] = input_content
                 data["ToolOutputs"] = output_content
 
+            # Create the escalation task via API
+            client = UiPath()
+            created_task = await client.tasks.create_async(
+                title="Agents Guardrail Task",
+                data=data,
+                app_name=self.app_name,
+                app_folder_path=self.app_folder_path,
+                recipient=task_recipient,
+            )
+
+            # Store task URL in metadata for observability — before interrupt
+            task_id = created_task.id
+            task_url = f"{UiPathConfig.base_url}/actions_/tasks/{task_id}"
+            metadata["escalation_data"]["task_url"] = task_url
+            metadata["node_type"] = "create_hitl_task"
+
+            # Persist task info in state so the interrupt node can read it,
+            # and so that on resume this node can detect it already ran.
+            return {
+                "inner_state": {
+                    "hitl_task_info": {
+                        create_task_node_name: created_task.model_dump(mode="json")
+                    },
+                }
+            }
+
+        _create_task_node.__metadata__ = metadata  # type: ignore[attr-defined]
+
+        # ── Node 2: interrupt and wait for escalation result ─────────────
+        async def _interrupt_node(
+            state: AgentGuardrailsGraphState,
+        ) -> Dict[str, Any] | Command[Any]:
+            hitl_info = state.inner_state.hitl_task_info
+            task_info = hitl_info.get(create_task_node_name) if hitl_info else None
+            if task_info is None:
+                raise AgentTerminationException(
+                    code=UiPathErrorCode.EXECUTION_ERROR,
+                    title="Escalation task not found",
+                    detail="The escalation task was not found in state. "
+                    "The create-task node must run before the interrupt node.",
+                )
+
+            created_task = Task.model_validate(task_info)
+            task_recipient = (
+                TaskRecipient.model_validate(task_info["recipient"])
+                if task_info.get("recipient")
+                else None
+            )
+
             escalation_result = interrupt(
-                CreateEscalation(
+                WaitEscalation(
+                    action=created_task,
                     app_name=self.app_name,
                     app_folder_path=self.app_folder_path,
-                    title="Agents Guardrail Task",
-                    data=data,
                     recipient=task_recipient,
                 )
             )
@@ -214,24 +275,49 @@ async def _node(
                 if reviewed_by:
                     metadata["escalation_data"]["reviewed_by"] = reviewed_by
 
+            # Clear the stored task data
+            cleanup_update: Dict[str, Any] = {
+                "inner_state": {"hitl_task_info": {create_task_node_name: None}},
+            }
+
             if escalation_result.action == "Approve":
-                return _process_escalation_response(
+                result = _process_escalation_response(
                     state,
                     escalation_result.data,
                     scope,
                     execution_stage,
                     guarded_component_name,
                 )
+                # Merge the cleanup into the response
+                if isinstance(result, Command) and result.update:
+                    _deep_merge(result.update, cleanup_update)
+                elif isinstance(result, dict):
+                    _deep_merge(result, cleanup_update)
+                else:
+                    result = cleanup_update
+                return result
 
             raise AgentTerminationException(
                 code=UiPathErrorCode.EXECUTION_ERROR,
                 title="Escalation rejected",
-                detail=f"Please contact your administrator. Action was rejected after reviewing the task created by guardrail [{guardrail.name}], with reason: {escalation_result.data.get('Reason', None)}",
+                detail=f"Action was rejected after reviewing the task created by guardrail [{guardrail.name}], with reason: {escalation_result.data.get('Reason', None)}",
             )
 
-        _node.__metadata__ = metadata  # type: ignore[attr-defined]
+        _interrupt_node.__metadata__ = metadata  # type: ignore[attr-defined]
 
-        return node_name, _node
+        return [
+            (create_task_node_name, _create_task_node),
+            (interrupt_node_name, _interrupt_node),
+        ]
+
+
+def _deep_merge(target: Dict[str, Any], source: Dict[str, Any]) -> None:
+    """Recursively merge *source* into *target* in place (dict values are merged, others overwritten)."""
+    for key, value in source.items():
+        if key in target and isinstance(target[key], dict) and isinstance(value, dict):
+            _deep_merge(target[key], value)
+        else:
+            target[key] = value
 
 
 def _parse_reviewed_data(value: Any) -> Any:

src/uipath_langchain/agent/react/guardrails/guardrails_subgraph.py

@@ -15,6 +15,7 @@
 from uipath_langchain.agent.guardrails.actions.base_action import (
     GuardrailAction,
     GuardrailActionNode,
+    GuardrailActionNodes,
 )
 from uipath_langchain.agent.guardrails.guardrail_nodes import (
     create_agent_init_guardrail_node,
@@ -172,16 +173,25 @@ def _build_guardrail_node_chain(
     guardrail, action = guardrails[-1]
     remaining_guardrails = guardrails[:-1]
 
-    fail_node_name, fail_node = action.action_node(
+    action_nodes: GuardrailActionNodes = action.action_node(
         guardrail=guardrail,
         scope=scope,
         execution_stage=execution_stage,
         guarded_component_name=guarded_node_name,
     )
 
+    # Normalize to a list: single tuple → list of one tuple
+    if isinstance(action_nodes, tuple):
+        action_nodes = [action_nodes]
+
+    # The first node is where the guardrail-eval routes on failure;
+    # the last node gets the outgoing edge to ``next_node``.
+    first_fail_node_name = action_nodes[0][0]
+    last_fail_node_name = action_nodes[-1][0]
+
     # Create the guardrail evaluation node.
     guardrail_node_name, guardrail_node = node_factory(
-        guardrail, execution_stage, next_node, fail_node_name
+        guardrail, execution_stage, next_node, first_fail_node_name
     )
 
     guardrail_node_metadata = getattr(guardrail_node, "__metadata__", None) or {}
@@ -191,21 +201,27 @@ def _build_guardrail_node_chain(
         "node_type": "guardrail_evaluation",
     }
 
-    fail_node_metadata = getattr(fail_node, "__metadata__", None) or {}
-    fail_node_metadata = {
-        **fail_node_metadata,
-        "action_type": action.action_type,
-        "node_type": "guardrail_action",
-        "tool_type": guardrail_node_metadata.get("tool_type"),
-    }
-
     subgraph.add_node(
         guardrail_node_name, guardrail_node, metadata={**guardrail_node_metadata}
     )
-    subgraph.add_node(fail_node_name, fail_node, metadata={**fail_node_metadata})
 
-    # Failure path route to the next node
-    subgraph.add_edge(fail_node_name, next_node)
+    # Add all action nodes and chain them together
+    for i, (fail_node_name, fail_node) in enumerate(action_nodes):
+        fail_node_metadata = getattr(fail_node, "__metadata__", None) or {}
+        fail_node_metadata = {
+            **fail_node_metadata,
+            "action_type": action.action_type,
+            "node_type": "guardrail_action",
+            "tool_type": guardrail_node_metadata.get("tool_type"),
+        }
+        subgraph.add_node(fail_node_name, fail_node, metadata={**fail_node_metadata})
+        # Chain consecutive action nodes: node[i] → node[i+1]
+        if i > 0:
+            prev_name = action_nodes[i - 1][0]
+            subgraph.add_edge(prev_name, fail_node_name)
+
+    # Last action node routes to the next guardrail / inner node
+    subgraph.add_edge(last_fail_node_name, next_node)
 
     previous_node_name = _build_guardrail_node_chain(
         subgraph,

src/uipath_langchain/agent/react/types.py

@@ -35,6 +35,7 @@ class InnerAgentGuardrailsGraphState(InnerAgentGraphState):
     guardrail_validation_result: Optional[bool] = None
     guardrail_validation_details: Optional[str] = None
     agent_result: Optional[dict[str, Any]] = None
+    hitl_task_info: Optional[Any] = {}
 
 
 class AgentGraphState(BaseModel):

tests/agent/guardrails/actions/test_block_action.py

@@ -73,4 +73,7 @@ async def test_node_name_and_exception(
             await node(AgentGuardrailsGraphState(messages=[]))
 
         # The exception string is the provided reason
-        assert str(excinfo.value) == "Sensitive data detected"
+        assert (
+            str(excinfo.value)
+            == "Execution was blocked by guardrail [My Guardrail v1], with reason: Sensitive data detected"
+        )