feat: JIT escalation apps support

src/uipath_langchain/agent/exceptions/exceptions.py

@@ -42,6 +42,9 @@ class AgentRuntimeErrorCode(str, Enum):
     TERMINATION_GUARDRAIL_ERROR = "TERMINATION_GUARDRAIL_ERROR"
     TERMINATION_ESCALATION_REJECTED = "TERMINATION_ESCALATION_REJECTED"
     TERMINATION_ESCALATION_ERROR = "TERMINATION_ESCALATION_ERROR"
+    ESCALATION_JIT_DEBUG_MISSING_PROJECT_KEY = (
+        "ESCALATION_JIT_DEBUG_MISSING_PROJECT_KEY"
+    )
 
     # State
     STATE_ERROR = "STATE_ERROR"

src/uipath_langchain/agent/tools/escalation_tool.py

@@ -19,7 +19,7 @@
 from uipath.eval.mocks import mockable
 from uipath.platform import UiPath
 from uipath.platform.action_center.tasks import Task, TaskRecipient
-from uipath.platform.common import WaitEscalation
+from uipath.platform.common import UiPathConfig, WaitEscalation
 from uipath.runtime.errors import UiPathErrorCategory
 
 from uipath_langchain._utils import (
@@ -203,6 +203,44 @@
     )
 
 
+async def _resolve_is_debug_run() -> bool:
+    """Determine whether the current run is a debug run.
+
+    Reads the running job key (``UIPATH_JOB_KEY``) from the runtime environment
+    via ``UiPathConfig``, then retrieves the job from Orchestrator (the SDK
+    builds the ``Jobs/...GetByKey`` URL and injects the ``x-uipath-folderkey``
+    header from ``UIPATH_FOLDER_KEY`` on every request from its HTTP client).
+
+    The job's ``ParentContext`` field is a JSON string such as
+    ``{"IsDebug": true}``; when ``IsDebug`` is truthy the run is a debug run.
+
+    Returns:
+        True when the current job is a debug run, False otherwise (including
+        when the job key is unavailable or the parent context cannot be read).
+    """
+    job_key = UiPathConfig.job_key
+    if not job_key:
+        return False
+
+    client = UiPath()
+    job = await client.jobs.retrieve_async(job_key=job_key)
+
+    parent_context_raw = job.parent_context
+    if not parent_context_raw:
+        return False
+
+    try:
+        parent_context = json.loads(parent_context_raw)
+    except json.JSONDecodeError:
+        _escalation_logger.warning(
+            "Unable to parse job ParentContext to determine debug run: %r",
+            parent_context_raw,
+        )
+        return False
+
+    return bool(parent_context.get("IsDebug"))
+
+
 async def create_task_for_channel(
     client: UiPath,
     channel: EscalationChannel,
@@ -211,6 +249,10 @@
     data: dict[str, Any],
     recipient: TaskRecipient | None,
     folder_path: str | None,
+    app_project_key: str | None = None,
+    app_type: str | None = None,
+    is_debug: bool = False,
+    action_schema: Any = None,
 ) -> Task:
     """Create the human task backing an escalation channel."""
     if isinstance(channel, AgentQuickFormEscalationChannel):
@@ -227,6 +269,7 @@
             labels=channel.labels,
             is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
             actionable_message_metadata=channel.properties.actionable_message_meta_data,
+            solution_id=channel.properties.solution_id,
         )
     return await client.tasks.create_async(
         title=title,
@@ -238,6 +281,10 @@
         labels=channel.labels,
         is_actionable_message_enabled=channel.properties.is_actionable_message_enabled,
         actionable_message_metadata=channel.properties.actionable_message_meta_data,
+        app_project_key=app_project_key,
+        app_type=app_type,
+        action_schema=action_schema,
+        solution_id=channel.properties.solution_id,
     )
 
 
@@ -311,6 +358,26 @@
                 else None
             )
 
+        is_debug = UiPathConfig.is_rooted_to_debug_job
+        if not is_debug:
+            is_debug = await _resolve_is_debug_run()
+        print('Resolved debug run', is_debug)
+        if is_debug:
+            app_project_key = channel.properties.project_key
+        app_version = channel.properties.app_version
+        action_schema = channel.properties.action_schema
+
+        if is_debug and app_version == 0 and not app_project_key:
+            raise AgentRuntimeError(
+                code=AgentRuntimeErrorCode.ESCALATION_JIT_DEBUG_MISSING_PROJECT_KEY,
+                title="Unable to create the escalation task in debug mode",
+                detail=(
+                    "The app project key is missing from the escalation resource "
+                    "configuration, so the app cannot be resolved in debug mode."
+                ),
+                category=UiPathErrorCategory.USER,
+            )
+
         task_title = "Escalation Task"
         if tool.metadata is not None:
             # Recipient requires runtime resolution, store in metadata after resolving
@@ -354,6 +421,10 @@
                     data=serialized_data,
                     recipient=recipient,
                     folder_path=folder_path,
+                    app_project_key=app_project_key,
+                    app_type=channel.properties.app_type,
+                    is_debug=is_debug,
+                    action_schema=action_schema,
                 )
 
                 if created_task.id is not None:

tests/agent/tools/test_escalation_tool.py

@@ -16,6 +16,7 @@
 )
 from uipath.platform.action_center.tasks import Task, TaskRecipient, TaskRecipientType
 
+from uipath_langchain.agent.exceptions import AgentRuntimeError
 from uipath_langchain.agent.tools.escalation_memory import (
     EscalationMemoryCachedResult,
 )
@@ -213,6 +214,88 @@ async def test_escalation_tool_metadata_recipient_none_when_no_recipients(
         assert tool.metadata is not None
         assert tool.metadata["recipient"] is None
 
+    @pytest.fixture
+    def escalation_resource_jit(self):
+        """Escalation resource carrying JIT (debug) project key and app type."""
+        return AgentEscalationResourceConfig(
+            name="approval",
+            description="Request approval",
+            channels=[
+                AgentEscalationChannel(
+                    name="action_center",
+                    type="actionCenter",
+                    description="Action Center channel",
+                    input_schema={"type": "object", "properties": {}},
+                    output_schema={"type": "object", "properties": {}},
+                    properties=AgentEscalationChannelProperties(
+                        app_name="ApprovalApp",
+                        app_version=1,
+                        resource_key="test-key",
+                        project_key="proj-key-abc",
+                        app_type="Custom",
+                    ),
+                    recipients=[
+                        StandardRecipient(
+                            type=AgentEscalationRecipientType.USER_EMAIL,
+                            value="user@example.com",
+                        )
+                    ],
+                )
+            ],
+        )
+
+    @pytest.mark.asyncio
+    @patch.dict(os.environ, {"UIPATH_PROJECT_ID": "proj-1"})
+    @patch("uipath_langchain.agent.tools.escalation_tool.UiPath")
+    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
+    async def test_escalation_tool_passes_jit_fields_in_debug(
+        self, mock_interrupt, mock_uipath_class, escalation_resource_jit
+    ):
+        """In a debug (studio) run, the project key, app type and is_debug flag are forwarded."""
+        mock_client = MagicMock()
+        mock_client.tasks.create_async = AsyncMock(return_value=_make_mock_task())
+        mock_uipath_class.return_value = mock_client
+
+        mock_result = MagicMock()
+        mock_result.action = None
+        mock_result.data = {}
+        mock_result.is_deleted = False
+        mock_interrupt.return_value = mock_result
+
+        tool = create_escalation_tool(escalation_resource_jit)
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+        await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
+
+        kwargs = mock_client.tasks.create_async.call_args.kwargs
+        assert kwargs["app_project_key"] == "proj-key-abc"
+        assert kwargs["app_type"] == "Custom"
+        assert kwargs["is_debug"] is True
+
+    @pytest.mark.asyncio
+    @patch.dict(os.environ, {"UIPATH_PROJECT_ID": "proj-1"})
+    @patch("uipath_langchain.agent.tools.escalation_tool.UiPath")
+    @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")
+    async def test_escalation_tool_raises_in_debug_without_project_key(
+        self, mock_interrupt, mock_uipath_class, escalation_resource
+    ):
+        """In a debug run the task cannot be created without a project key for the undeployed app."""
+        mock_client = MagicMock()
+        mock_client.tasks.create_async = AsyncMock(return_value=_make_mock_task())
+        mock_uipath_class.return_value = mock_client
+
+        mock_result = MagicMock()
+        mock_result.action = None
+        mock_result.data = {}
+        mock_result.is_deleted = False
+        mock_interrupt.return_value = mock_result
+
+        tool = create_escalation_tool(escalation_resource)
+        call = ToolCall(args={}, id="test-call", name=tool.name)
+
+        with pytest.raises(AgentRuntimeError):
+            await tool.awrapper(tool, call, {})  # type: ignore[attr-defined]
+        mock_client.tasks.create_async.assert_not_called()
+
     @pytest.mark.asyncio
     @patch("uipath_langchain.agent.tools.escalation_tool.UiPath")
     @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt")