From da79a744077a8aaaa26c06192a4c5f8f51208f5f Mon Sep 17 00:00:00 2001 From: Sandeepan-Ghosh-0312 Date: Tue, 21 Jul 2026 16:29:37 +0530 Subject: [PATCH] feat: JIT escalation apps support --- .../agent/exceptions/exceptions.py | 3 + .../guardrails/actions/escalate_action.py | 25 + .../agent/tools/escalation_jit.py | 387 ++++++++++ .../agent/tools/escalation_tool.py | 50 +- .../actions/test_escalate_action.py | 149 ++++ tests/agent/tools/test_escalation_tool.py | 715 ++++++++++++++++++ 6 files changed, 1324 insertions(+), 5 deletions(-) create mode 100644 src/uipath_langchain/agent/tools/escalation_jit.py diff --git a/src/uipath_langchain/agent/exceptions/exceptions.py b/src/uipath_langchain/agent/exceptions/exceptions.py index 65ebbaf81..266213e43 100644 --- a/src/uipath_langchain/agent/exceptions/exceptions.py +++ b/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_APP_JIT_DEBUG_MISSING_INFORMATION = ( + "ESCALATION_APP_JIT_DEBUG_MISSING_INFORMATION" + ) # State STATE_ERROR = "STATE_ERROR" diff --git a/src/uipath_langchain/agent/guardrails/actions/escalate_action.py b/src/uipath_langchain/agent/guardrails/actions/escalate_action.py index 65ae36d2b..966acb256 100644 --- a/src/uipath_langchain/agent/guardrails/actions/escalate_action.py +++ b/src/uipath_langchain/agent/guardrails/actions/escalate_action.py @@ -34,6 +34,10 @@ from ...messages.message_utils import replace_tool_calls from ...react.types import AgentGuardrailsGraphState from ...react.utils import extract_current_tool_call_index, find_latest_ai_message +from ...tools.escalation_jit import ( + resolve_is_debug_run_safely, + resolve_jit_escalation_app, +) from ...tools.escalation_recipient import resolve_recipient_value from ..types import ExecutionStage from ..utils import _extract_tool_args_from_message, get_message_content @@ -211,6 +215,24 @@ async def _create_task_node( data["ToolInputs"] = input_content data["ToolOutputs"] = output_content + # Resolve inline (JIT) app targeting for a not-yet-deployed app in a + # debug run, mirroring the escalation tool's inline-app flow. The + # guardrail escalate app config carries no app_type/action_schema, so + # both are resolved from the solution at runtime when applicable. + is_debug = await resolve_is_debug_run_safely() + ( + app_project_key, + app_type, + action_schema, + ) = await resolve_jit_escalation_app( + app_name=self.app_name, + app_version=self.version, + app_type=None, + action_schema=None, + folder_path=self.app_folder_path, + is_debug=is_debug, + ) + # Create the escalation task via API client = UiPath() created_task = await client.tasks.create_async( @@ -219,6 +241,9 @@ async def _create_task_node( app_name=self.app_name, app_folder_path=self.app_folder_path, recipient=task_recipient, + app_project_key=app_project_key, + app_type=app_type, + action_schema=action_schema, ) # Store task URL in metadata for observability — before interrupt diff --git a/src/uipath_langchain/agent/tools/escalation_jit.py b/src/uipath_langchain/agent/tools/escalation_jit.py new file mode 100644 index 000000000..93996dac3 --- /dev/null +++ b/src/uipath_langchain/agent/tools/escalation_jit.py @@ -0,0 +1,387 @@ +"""Just-in-time (JIT) escalation app resolution for Action Center integration. + +An escalation may target an app that has not been deployed yet — an *inline* +app that lives only inside the running agent's solution. During a debug run the +app project (``designId``), its ``app_type`` and, for low-code apps, its action +schema must be resolved from the Studio backend at runtime so the human task can +be created without a deployed app. + +This module holds that resolution so both the escalation *tool* (factory-path, +``escalation_tool.py``) and the escalate *guardrail action* +(``guardrails/actions/escalate_action.py``) can share one implementation. +""" + +import asyncio +import json +import logging +from typing import Any + +from uipath.core.feature_flags import FeatureFlags +from uipath.platform import UiPath +from uipath.platform.action_center.tasks import is_low_code_app +from uipath.platform.common import UiPathConfig +from uipath.platform.common._bindings import _resource_overwrites +from uipath.runtime.errors import UiPathErrorCategory + +from ..exceptions import AgentRuntimeError, AgentRuntimeErrorCode + +_escalation_jit_logger = logging.getLogger(__name__) + +_JIT_ESCALATION_APPS_FEATURE_FLAG = "EnableJITEscalationApps" +_IS_DEBUG_RESOLVE_TIMEOUT_SECONDS = 15 + + +def _has_app_name_override(app_name: str | None, folder_path: str | None) -> bool: + """Return whether a bindings resource overwrite exists for the app. + + Used to disambiguate an ``app_version`` of 1 for backward compatibility: a + genuinely deployed app carries an app resource binding, whereas an inline + app (formerly sent from the frontend with a buggy version of 1) does not. + + Mirrors the key resolution used by the ``@resource_override`` decorator on + ``tasks.create_async`` (``resource_type="app"``, + ``resource_identifier="app_name"``, ``folder_identifier="app_folder_path"``): + it looks up ``app.{app_name}`` and prefers the folder-qualified + ``app.{app_name}.{folder_path}`` when that fuller key is present. + + Args: + app_name: The escalation channel's design-time app name. + folder_path: The app folder path, used to disambiguate the overwrite key. + + Returns: + True when a matching overwrite is present in the current + :data:`_resource_overwrites` context, False otherwise. + """ + if not app_name: + return False + + overwrites = _resource_overwrites.get() + if not overwrites: + return False + + key = f"app.{app_name}" + if folder_path and f"{key}.{folder_path}" in overwrites: + key = f"{key}.{folder_path}" + + overwrite = overwrites.get(key) + if overwrite is None: + return False + + return True + + +def _is_inline_app( + app_type: str | None, + app_version: Any, + app_name: str | None, + folder_path: str | None, +) -> bool: + """Return True when the escalation targets an inline (not-yet-deployed) app. + + An inline app is identified by ``app_version == 0``. Backward compatibility: + older frontends sent version 1 for inline apps by mistake; a real deployed + app referenced with version 1 carries an app resource binding, so when no + binding override exists a version-1 app is treated as inline too. + """ + if not (is_low_code_app(app_type) or app_type is None): + return False + if app_version == 0: + return True + return app_version == 1 and not _has_app_name_override(app_name, folder_path) + + +def _app_type_from_project_type(project_type: str | None) -> str | None: + """Map a solution project's ``projectType`` to a task ``app_type``. + + The Solution backend reports ``AppV2`` for a coded app and ``Process`` for a + low code app, whereas task creation expects ``Coded`` / ``Custom``. Returns + None for an unrecognized or missing ``projectType``. + """ + return {"AppV2": "Coded", "Process": "Custom"}.get(project_type or "") + + +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. + + On a successful resolution to a debug run, records it on + ``UiPathConfig.is_rooted_to_debug_job`` so downstream task creation picks it + up. + + 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() + async with asyncio.timeout(_IS_DEBUG_RESOLVE_TIMEOUT_SECONDS): + 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_jit_logger.warning( + "Unable to parse job ParentContext to determine debug run: %r", + parent_context_raw, + ) + return False + + is_debug = bool(parent_context.get("IsDebug")) + if is_debug: + UiPathConfig.is_rooted_to_debug_job = True + return is_debug + + +async def _resolve_solution_id(client: UiPath) -> str | None: + """Return the current solution id, resolved lazily from the project id. + + Prefers the value cached on ``UiPathConfig`` (populated when the debug + runtime applies resource overwrites before the agent runs). Falls back to + querying the Studio project endpoint directly so escalation still works when + no resource overwrites were loaded. + + Returns: + The solution id, or None when it cannot be resolved. + """ + solution_id = UiPathConfig.studio_solution_id + if solution_id: + return solution_id + + project_id = UiPathConfig.project_id + if not project_id: + return None + + response = await client.api_client.request_async( + "GET", + url=f"/studio_/backend/api/Project/{project_id}", + scoped="org", + ) + solution_id = response.json().get("solutionId") + UiPathConfig.studio_solution_id = solution_id + return solution_id + + +async def _resolve_app_project( + client: UiPath, app_name: str | None +) -> dict[str, Any] | None: + """Resolve a not-yet-deployed app's project from the solution at runtime. + + Looks up the solution the running agent belongs to and returns the app + project (``isApp`` true or ``projectType`` is AppV2) whose ``name`` matches ``app_name``. + + Args: + client: The UiPath SDK client used to call the Studio backend. + app_name: The escalation channel's app name, used to disambiguate when a + solution contains more than one app. + + Returns: + The app project dict, or None when no + app can be resolved. + """ + solution_id = await _resolve_solution_id(client) + if not solution_id: + return None + + response = await client.api_client.request_async( + "GET", + url=f"/studio_/backend/api/Solution/{solution_id}", + scoped="org", + ) + apps = [ + project + for project in response.json().get("projects", []) + if project.get("isApp") or project.get("projectType") == "AppV2" + ] + + if app_name is not None: + for app in apps: + if app.get("name") == app_name: + return app + + return None + + +def _find_schema_file_id(node: Any) -> str | None: + """Depth-first search the FileOperations structure for the action schema file. + + The app project stores its action schema as ``schemas/schema-.json``. + Returns the content ``id`` of the first file whose name matches that pattern. + """ + if not isinstance(node, dict): + return None + + for file in node.get("files", []): + name = file.get("name", "") + if name.startswith("schema-") and name.endswith(".json"): + return file.get("id") + + for folder in node.get("folders", []): + found = _find_schema_file_id(folder) + if found is not None: + return found + + return None + + +async def _resolve_app_action_schema( + client: UiPath, app_project_id: str +) -> dict[str, Any] | None: + """Fetch a not-yet-deployed app's action schema from the Studio backend. + + Walks the app project's file structure to locate ``schemas/schema-*.json``, + then reads that file. The returned object mirrors the ``actionSchema`` the + deployed-apps path returns (``key``, ``inOuts``, ``inputs``, ``outputs``, + ``outcomes``), so the JIT (debug) task can be created without a deployed app. + + Args: + client: The UiPath SDK client used to call the Studio backend. + app_project_id: The app project's ``id`` (not its ``designId``). + + Returns: + The action schema dict, or None when the schema file cannot be found. + """ + structure = ( + await client.api_client.request_async( + "GET", + url=f"/studio_/backend/api/Project/{app_project_id}/FileOperations/Structure", + scoped="org", + ) + ).json() + + file_id = _find_schema_file_id(structure) + if not file_id: + return None + + return ( + await client.api_client.request_async( + "GET", + url=f"/studio_/backend/api/Project/{app_project_id}/FileOperations/File/{file_id}", + scoped="org", + ) + ).json() + + +async def resolve_is_debug_run_safely() -> bool: + """Return whether the current run is a debug run, tolerating failures. + + Prefers the flag already recorded on ``UiPathConfig.is_rooted_to_debug_job`` + and otherwise probes Orchestrator, falling back to release mode on any error. + Mirrors the unconditional debug resolution performed before an escalation + task is created (a successful probe also records the flag on + ``UiPathConfig`` for downstream task creation). + + Returns: + True when the current job is a debug run, False otherwise. + """ + is_debug = UiPathConfig.is_rooted_to_debug_job + if not is_debug: + try: + is_debug = await _resolve_is_debug_run() + except Exception: + # fallback to release mode + is_debug = False + return is_debug + + +async def resolve_jit_escalation_app( + *, + app_name: str | None, + app_version: Any, + app_type: str | None, + action_schema: Any, + folder_path: str | None, + is_debug: bool, +) -> tuple[str | None, str | None, Any]: + """Resolve inline (JIT) app targeting info for an escalation task. + + When the escalation targets an inline (not-yet-deployed) app in a debug run + and the JIT feature flag is enabled, resolves the app project (``designId``), + its ``app_type`` and — for low-code apps — its action schema from the + solution at runtime so the human task can be created without a deployed app. + + Args: + app_name: The design-time app name of the escalation target. + app_version: The escalation app version (0, or 1 for legacy inline apps). + app_type: The app type when already known (``Coded`` / ``Custom``), else + None to resolve from the project's ``projectType``. + action_schema: The action schema when already known, else None to resolve + from the Studio backend for low-code apps. + folder_path: The app folder path, used to disambiguate the overwrite key. + is_debug: Whether the current run is a debug run. + + Returns: + The tuple ``(app_project_key, app_type, action_schema)``. + ``app_project_key`` is None unless the escalation targets an inline app + that was resolved from the solution at runtime; ``app_type`` and + ``action_schema`` are returned unchanged when no resolution applies. + + Raises: + AgentRuntimeError: When a low-code inline app is targeted in debug mode + but its project key or action schema could not be resolved. + """ + jit_enabled = FeatureFlags.is_flag_enabled( + _JIT_ESCALATION_APPS_FEATURE_FLAG, default=False + ) + is_inline_jit_app = jit_enabled and _is_inline_app( + app_type, app_version, app_name, folder_path + ) + if not (is_inline_jit_app and is_debug): + return None, app_type, action_schema + + app_project_key: str | None = None + try: + app_project = await _resolve_app_project(UiPath(), app_name) + if app_project is not None: + app_project_key = app_project.get("designId") + if not app_type: + app_type = _app_type_from_project_type(app_project.get("projectType")) + if ( + is_low_code_app(app_type) + and action_schema is None + and app_project.get("id") + ): + action_schema = await _resolve_app_action_schema( + UiPath(), app_project["id"] + ) + except Exception: + _escalation_jit_logger.exception( + "Failed to resolve inline app '%s' from the solution at debug runtime", + app_name, + ) + + missing_fields = [ + label + for label, value in ( + ("app project key", app_project_key), + ("action schema", action_schema), + ) + if value is None + ] + if is_low_code_app(app_type) and missing_fields: + raise AgentRuntimeError( + code=AgentRuntimeErrorCode.ESCALATION_APP_JIT_DEBUG_MISSING_INFORMATION, + title="Unable to create the escalation in debug mode", + detail=( + f"Could not resolve the {', '.join(missing_fields)} " + f"for the app '{app_name}' from the solution at runtime, so the " + "app cannot be targeted in debug mode. Please open the agent " + "project and try again" + ), + category=UiPathErrorCategory.USER, + ) + + return app_project_key, app_type, action_schema diff --git a/src/uipath_langchain/agent/tools/escalation_tool.py b/src/uipath_langchain/agent/tools/escalation_tool.py index a6cf27923..d6fc272fc 100644 --- a/src/uipath_langchain/agent/tools/escalation_tool.py +++ b/src/uipath_langchain/agent/tools/escalation_tool.py @@ -18,7 +18,10 @@ ) from uipath.eval.mocks import mockable from uipath.platform import UiPath -from uipath.platform.action_center.tasks import Task, TaskRecipient +from uipath.platform.action_center.tasks import ( + Task, + TaskRecipient, +) from uipath.platform.common import WaitEscalation from uipath.runtime.errors import UiPathErrorCategory @@ -42,6 +45,10 @@ AgentStartupErrorCode, ) from ..react.types import AgentGraphState +from .escalation_jit import ( + resolve_is_debug_run_safely, + resolve_jit_escalation_app, +) from .escalation_memory import ( EscalationMemorySettings, _check_escalation_memory_cache, @@ -195,9 +202,14 @@ def _get_exported_trace_id(trace_id: str | None) -> str | None: return trace_id -def _try_get_channel_app_name(channel: EscalationChannel) -> str | None: +def _channel_app_prop(channel: EscalationChannel, prop: str) -> Any: + """Return an app property from an ``AgentEscalationChannel``, else None. + + Covers ``app_name`` / ``app_version`` / ``app_type``; other channel types + (e.g. quick form) carry no app properties, so None is returned. + """ return ( - channel.properties.app_name + getattr(channel.properties, prop, None) if isinstance(channel, AgentEscalationChannel) else None ) @@ -211,6 +223,9 @@ async def create_task_for_channel( data: dict[str, Any], recipient: TaskRecipient | None, folder_path: str | None, + app_project_key: str | None = None, + app_type: str | None = None, + action_schema: Any = None, ) -> Task: """Create the human task backing an escalation channel.""" if isinstance(channel, AgentQuickFormEscalationChannel): @@ -238,6 +253,9 @@ async def create_task_for_channel( 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, ) @@ -311,6 +329,25 @@ async def escalation_tool_fn(**kwargs: Any) -> dict[str, Any]: else None ) + app_project_key: str | None = None + app_version = _channel_app_prop(channel, "app_version") + app_name = _channel_app_prop(channel, "app_name") + app_type = _channel_app_prop(channel, "app_type") + action_schema = _channel_app_prop(channel, "action_schema") + + # Resolve the debug status unconditionally for all escalations + is_debug = await resolve_is_debug_run_safely() + + if isinstance(channel, AgentEscalationChannel): + app_project_key, app_type, action_schema = await resolve_jit_escalation_app( + app_name=app_name, + app_version=app_version, + app_type=app_type, + action_schema=action_schema, + folder_path=folder_path, + is_debug=is_debug, + ) + task_title = "Escalation Task" if tool.metadata is not None: # Recipient requires runtime resolution, store in metadata after resolving @@ -354,6 +391,9 @@ async def create_escalation_task(): data=serialized_data, recipient=recipient, folder_path=folder_path, + app_project_key=app_project_key, + app_type=app_type, + action_schema=action_schema, ) if created_task.id is not None: @@ -362,7 +402,7 @@ async def create_escalation_task(): return WaitEscalation( action=created_task, app_folder_path=folder_path, - app_name=_try_get_channel_app_name(channel), + app_name=_channel_app_prop(channel, "app_name"), recipient=recipient, ) @@ -507,7 +547,7 @@ async def escalation_wrapper( argument_properties=channel.argument_properties, metadata={ "tool_type": "escalation", - "display_name": _try_get_channel_app_name(channel) or channel.name, + "display_name": _channel_app_prop(channel, "app_name") or channel.name, "channel_type": channel.type, "recipient": None, "args_schema": input_model, diff --git a/tests/agent/guardrails/actions/test_escalate_action.py b/tests/agent/guardrails/actions/test_escalate_action.py index 9e1c03f9b..699ecce38 100644 --- a/tests/agent/guardrails/actions/test_escalate_action.py +++ b/tests/agent/guardrails/actions/test_escalate_action.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os from typing import Any, Dict from unittest.mock import AsyncMock, MagicMock, patch @@ -2293,3 +2294,151 @@ def test_parse_reviewed_data_list_passthrough(self): data = [1, 2, 3] result = _parse_reviewed_data(data) assert result is data + + +class TestEscalateActionJit: + """Just-in-time (JIT) inline-app resolution in the create-task node. + + Mirrors the escalation tool's inline-app flow: for a not-yet-deployed app in + a debug run with the feature flag on, the create-task node resolves the app + project key, app type and action schema at runtime and forwards them to + ``tasks.create_async``. + """ + + def _make_jit_action(self): + return EscalateAction( + app_name="ApprovalApp", + app_folder_path="Shared", + version=0, # inline (not-yet-deployed) app + recipient=DEFAULT_RECIPIENT, + ) + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_FEATURE_EnableJITEscalationApps": "true"}) + @patch("uipath_langchain.agent.guardrails.actions.escalate_action.UiPathConfig") + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_action_schema", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + @patch("uipath_langchain.agent.guardrails.actions.escalate_action.UiPath") + @patch( + "uipath_langchain.agent.guardrails.actions.escalate_action.resolve_recipient_value" + ) + async def test_create_task_node_resolves_jit_fields_in_debug( + self, + mock_resolve_recipient, + mock_uipath_class, + mock_jit_uipath_class, + mock_resolve_debug, + mock_resolve_project, + mock_resolve_schema, + mock_config, + ) -> None: + """In a debug run with the flag on, the app project key, app type and + action schema are resolved at runtime and forwarded to create_async.""" + mock_resolve_recipient.return_value = TaskRecipient( + value="test@example.com", type=TaskRecipientType.EMAIL + ) + mock_config.base_url = None + mock_config.tenant_name = "TestTenant" + + mock_task = _make_mock_task(recipient=MOCK_TASK_RECIPIENT) + mock_client = MagicMock() + mock_client.tasks.create_async = AsyncMock(return_value=mock_task) + mock_uipath_class.return_value = mock_client + + mock_resolve_debug.return_value = True + mock_resolve_project.return_value = { + "designId": "proj-key-abc", + "id": "proj-id", + "projectType": "Process", # -> Custom (low-code) + } + schema = { + "key": "schema-key", + "inOuts": [], + "inputs": [], + "outputs": [], + "outcomes": [], + } + mock_resolve_schema.return_value = schema + + action = self._make_jit_action() + guardrail = _make_default_guardrail() + create_task_name, create_task_fn, _, _ = _get_action_nodes( + action, guardrail, GuardrailScope.LLM, ExecutionStage.PRE_EXECUTION + ) + state = AgentGuardrailsGraphState( + messages=[HumanMessage(content="Test message")], + inner_state=InnerAgentGuardrailsGraphState( + guardrail_validation_details="Validation failed" + ), + ) + + await create_task_fn(state) + + call_kwargs = mock_client.tasks.create_async.call_args[1] + assert call_kwargs["app_project_key"] == "proj-key-abc" + assert call_kwargs["app_type"] == "Custom" + assert call_kwargs["action_schema"] == schema + + @pytest.mark.asyncio + @patch("uipath_langchain.agent.guardrails.actions.escalate_action.UiPathConfig") + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.guardrails.actions.escalate_action.UiPath") + @patch( + "uipath_langchain.agent.guardrails.actions.escalate_action.resolve_recipient_value" + ) + async def test_create_task_node_skips_jit_when_flag_disabled( + self, + mock_resolve_recipient, + mock_uipath_class, + mock_resolve_debug, + mock_config, + ) -> None: + """With the feature flag off, no JIT resolution happens and the JIT + fields are forwarded as None (a deployed app is targeted by name).""" + os.environ.pop("UIPATH_FEATURE_EnableJITEscalationApps", None) + mock_resolve_recipient.return_value = TaskRecipient( + value="test@example.com", type=TaskRecipientType.EMAIL + ) + mock_config.base_url = None + mock_config.tenant_name = "TestTenant" + + mock_task = _make_mock_task(recipient=MOCK_TASK_RECIPIENT) + mock_client = MagicMock() + mock_client.tasks.create_async = AsyncMock(return_value=mock_task) + mock_uipath_class.return_value = mock_client + + # Even if the run were debug, the disabled flag short-circuits JIT. + mock_resolve_debug.return_value = True + + action = self._make_jit_action() + guardrail = _make_default_guardrail() + create_task_name, create_task_fn, _, _ = _get_action_nodes( + action, guardrail, GuardrailScope.LLM, ExecutionStage.PRE_EXECUTION + ) + state = AgentGuardrailsGraphState( + messages=[HumanMessage(content="Test message")], + inner_state=InnerAgentGuardrailsGraphState( + guardrail_validation_details="Validation failed" + ), + ) + + await create_task_fn(state) + + call_kwargs = mock_client.tasks.create_async.call_args[1] + assert call_kwargs["app_project_key"] is None + assert call_kwargs["app_type"] is None + assert call_kwargs["action_schema"] is None diff --git a/tests/agent/tools/test_escalation_tool.py b/tests/agent/tools/test_escalation_tool.py index 84cdc99b1..03f372004 100644 --- a/tests/agent/tools/test_escalation_tool.py +++ b/tests/agent/tools/test_escalation_tool.py @@ -16,11 +16,23 @@ ) from uipath.platform.action_center.tasks import Task, TaskRecipient, TaskRecipientType +from uipath_langchain.agent.exceptions import AgentRuntimeError +from uipath_langchain.agent.tools.escalation_jit import ( + _app_type_from_project_type, + _find_schema_file_id, + _has_app_name_override, + _is_inline_app, + _resolve_app_action_schema, + _resolve_app_project, + _resolve_is_debug_run, + _resolve_solution_id, +) from uipath_langchain.agent.tools.escalation_memory import ( EscalationMemoryCachedResult, ) from uipath_langchain.agent.tools.escalation_tool import ( _build_escalation_memory_payload, + _channel_app_prop, _parse_task_data, create_escalation_tool, ) @@ -213,6 +225,200 @@ 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", + "UIPATH_FEATURE_EnableJITEscalationApps": "true", + }, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_action_schema", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_escalation_tool_resolves_jit_fields_in_debug( + self, + mock_interrupt, + mock_uipath_class, + mock_jit_uipath_class, + mock_resolve_debug, + mock_resolve_project, + mock_resolve_schema, + escalation_resource_jit, + ): + """In a debug run with the flag on, the app project key, app type and action + schema are resolved at runtime and forwarded; is_debug is NOT passed.""" + 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 + + mock_resolve_debug.return_value = True + mock_resolve_project.return_value = { + "designId": "proj-key-abc", + "id": "proj-id", + "projectType": "Process", # -> Custom + } + schema = { + "key": "schema-key", + "inOuts": [], + "inputs": [], + "outputs": [], + "outcomes": [], + } + mock_resolve_schema.return_value = schema + + 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["action_schema"] == schema + # is_debug is sourced from UiPathConfig inside the SDK, never passed here. + assert "is_debug" not in kwargs + + @pytest.mark.asyncio + @patch.dict( + os.environ, + { + "UIPATH_PROJECT_ID": "proj-1", + "UIPATH_FEATURE_EnableJITEscalationApps": "true", + }, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_escalation_tool_raises_in_debug_when_app_unresolvable( + self, + mock_interrupt, + mock_uipath_class, + mock_resolve_debug, + mock_resolve_project, + escalation_resource_jit, + ): + """A low-code app in debug that can't be resolved from the solution raises a + USER error and does not create a task.""" + 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 + + mock_resolve_debug.return_value = True + mock_resolve_project.return_value = None # cannot resolve the app + + tool = create_escalation_tool(escalation_resource_jit) + 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.dict(os.environ, {"UIPATH_PROJECT_ID": "proj-1"}, clear=False) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_escalation_tool_skips_jit_when_flag_disabled( + self, + mock_interrupt, + mock_uipath_class, + mock_resolve_debug, + mock_resolve_project, + escalation_resource_jit, + ): + """With the feature flag off, no JIT resolution happens even in debug.""" + monkeypatch_env = os.environ.pop("UIPATH_FEATURE_EnableJITEscalationApps", None) + assert monkeypatch_env is None # flag not set → disabled + + 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 + + mock_resolve_debug.return_value = True + + 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] + + # JIT app resolution is never attempted when the flag is off. + mock_resolve_project.assert_not_called() + kwargs = mock_client.tasks.create_async.call_args.kwargs + assert kwargs["app_project_key"] is None + @pytest.mark.asyncio @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") @@ -1366,3 +1572,512 @@ async def test_action_center_channel_does_not_dispatch_to_quickform( mock_client.tasks.create_async.assert_called_once() mock_client.tasks.create_quickform_async.assert_not_called() + + +def _app_channel(**props): + """Build an AgentEscalationChannel with the given app properties.""" + return AgentEscalationChannel( + name="action_center", + type="actionCenter", + description="Action Center channel", + input_schema={"type": "object", "properties": {}}, + output_schema={"type": "object", "properties": {}}, + properties=AgentEscalationChannelProperties( + app_name=props.get("app_name", "ApprovalApp"), + app_version=props.get("app_version", 0), + resource_key="test-key", + app_type=props.get("app_type"), + ), + recipients=[], + ) + + +class TestAppTypeFromProjectType: + @pytest.mark.parametrize( + "project_type,expected", + [ + ("AppV2", "Coded"), + ("Process", "Custom"), + (None, None), + ("", None), + ("Unknown", None), + ], + ) + def test_mapping(self, project_type, expected): + assert _app_type_from_project_type(project_type) == expected + + +class TestChannelAppProp: + def test_reads_property_from_agent_channel(self): + channel = _app_channel(app_name="MyApp", app_version=2, app_type="Coded") + assert _channel_app_prop(channel, "app_name") == "MyApp" + assert _channel_app_prop(channel, "app_version") == 2 + assert _channel_app_prop(channel, "app_type") == "Coded" + + def test_returns_none_for_non_agent_channel(self): + # A non-AgentEscalationChannel object carries no app properties. + assert _channel_app_prop(object(), "app_name") is None + + +class TestFindSchemaFileId: + def test_finds_schema_file_at_top_level(self): + node = {"files": [{"name": "schema-abc.json", "id": "file-1"}], "folders": []} + assert _find_schema_file_id(node) == "file-1" + + def test_finds_schema_file_in_nested_folder(self): + node = { + "files": [{"name": "other.json", "id": "x"}], + "folders": [ + {"files": [{"name": "schema-xyz.json", "id": "file-2"}], "folders": []} + ], + } + assert _find_schema_file_id(node) == "file-2" + + def test_returns_none_when_no_schema_file(self): + node = {"files": [{"name": "data.json", "id": "x"}], "folders": []} + assert _find_schema_file_id(node) is None + + def test_returns_none_for_non_dict(self): + assert _find_schema_file_id(None) is None + assert _find_schema_file_id("not-a-node") is None + + +class TestHasAppNameOverride: + def _set_overwrites(self, overwrites): + from uipath.platform.common._bindings import _resource_overwrites + + return _resource_overwrites, _resource_overwrites.set(overwrites) + + def test_false_when_no_app_name(self): + assert _has_app_name_override(None, None) is False + + def test_false_when_no_overwrites_context(self): + assert _has_app_name_override("MyApp", None) is False + + def test_true_when_app_key_present(self): + var, token = self._set_overwrites({"app.MyApp": object()}) + try: + assert _has_app_name_override("MyApp", None) is True + finally: + var.reset(token) + + def test_true_when_folder_qualified_key_present(self): + var, token = self._set_overwrites({"app.MyApp.Shared": object()}) + try: + assert _has_app_name_override("MyApp", "Shared") is True + finally: + var.reset(token) + + def test_false_when_key_absent(self): + var, token = self._set_overwrites({"app.OtherApp": object()}) + try: + assert _has_app_name_override("MyApp", None) is False + finally: + var.reset(token) + + +class TestIsInlineApp: + @pytest.mark.parametrize( + "app_type,app_version,expected", + [ + ("Custom", 0, True), + (None, 0, True), + ("Custom", 1, True), # v1 low-code, no binding override -> inline + ("Coded", 0, False), # coded is never inline + ("Custom", 2, False), # only v0/v1 qualify + ], + ) + def test_predicate(self, app_type, app_version, expected): + assert _is_inline_app(app_type, app_version, "MyApp", None) is expected + + def test_v1_with_binding_override_is_not_inline(self): + from uipath.platform.common._bindings import _resource_overwrites + + token = _resource_overwrites.set({"app.MyApp": object()}) + try: + assert _is_inline_app("Custom", 1, "MyApp", None) is False + finally: + _resource_overwrites.reset(token) + + +class TestResolveIsDebugRun: + @pytest.mark.asyncio + async def test_returns_false_when_no_job_key(self, monkeypatch): + monkeypatch.delenv("UIPATH_JOB_KEY", raising=False) + assert await _resolve_is_debug_run() is False + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_JOB_KEY": "job-1"}) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + async def test_true_sets_config(self, mock_uipath_class): + from uipath.platform.common import UiPathConfig + + UiPathConfig.reset() + try: + job = MagicMock() + job.parent_context = '{"IsDebug": true}' + mock_client = MagicMock() + mock_client.jobs.retrieve_async = AsyncMock(return_value=job) + mock_uipath_class.return_value = mock_client + + result = await _resolve_is_debug_run() + + assert result is True + assert UiPathConfig.is_rooted_to_debug_job is True + finally: + UiPathConfig.reset() + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_JOB_KEY": "job-1"}) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + async def test_false_does_not_set_config(self, mock_uipath_class): + from uipath.platform.common import UiPathConfig + + UiPathConfig.reset() + try: + job = MagicMock() + job.parent_context = '{"IsDebug": false}' + mock_client = MagicMock() + mock_client.jobs.retrieve_async = AsyncMock(return_value=job) + mock_uipath_class.return_value = mock_client + + result = await _resolve_is_debug_run() + + assert result is False + assert UiPathConfig.is_rooted_to_debug_job is False + finally: + UiPathConfig.reset() + + +class TestJitResolutionHelpers: + """Runtime resolution of the app project / solution / action schema.""" + + @pytest.fixture(autouse=True) + def _reset_config(self): + from uipath.platform.common import UiPathConfig + + UiPathConfig.reset() + yield + UiPathConfig.reset() + + def _client_returning(self, by_url): + """MagicMock client whose api_client.request_async returns a canned + ``.json()`` payload chosen by substring match on the request url.""" + + def _side_effect(method, url=None, **kwargs): + for needle, payload in by_url.items(): + if needle in url: + resp = MagicMock() + resp.json.return_value = payload + return resp + raise AssertionError(f"unexpected url: {url}") + + client = MagicMock() + client.api_client.request_async = AsyncMock(side_effect=_side_effect) + return client + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_PROJECT_ID": "proj-1"}) + async def test_resolve_solution_id_from_project(self): + from uipath.platform.common import UiPathConfig + + client = self._client_returning({"/Project/proj-1": {"solutionId": "sol-9"}}) + assert await _resolve_solution_id(client) == "sol-9" + # Cached on the config for subsequent lookups. + assert UiPathConfig.studio_solution_id == "sol-9" + + @pytest.mark.asyncio + async def test_resolve_solution_id_none_without_project(self, monkeypatch): + monkeypatch.delenv("UIPATH_PROJECT_ID", raising=False) + client = MagicMock() + assert await _resolve_solution_id(client) is None + + @pytest.mark.asyncio + async def test_resolve_app_project_matches_by_name(self): + from uipath.platform.common import UiPathConfig + + UiPathConfig.studio_solution_id = "sol-9" + client = self._client_returning( + { + "/Solution/sol-9": { + "projects": [ + {"name": "Other", "isApp": True, "designId": "d0", "id": "i0"}, + { + "name": "ApprovalApp", + "isApp": True, + "designId": "d1", + "id": "i1", + "projectType": "Process", + }, + {"name": "NotAnApp", "isApp": False}, + ] + } + } + ) + app = await _resolve_app_project(client, "ApprovalApp") + assert app is not None + assert app["designId"] == "d1" + assert app["projectType"] == "Process" + + @pytest.mark.asyncio + async def test_resolve_app_project_none_when_no_match(self): + from uipath.platform.common import UiPathConfig + + UiPathConfig.studio_solution_id = "sol-9" + client = self._client_returning( + {"/Solution/sol-9": {"projects": [{"name": "X", "isApp": True}]}} + ) + assert await _resolve_app_project(client, "ApprovalApp") is None + + @pytest.mark.asyncio + async def test_resolve_app_project_none_without_solution(self, monkeypatch): + monkeypatch.delenv("UIPATH_PROJECT_ID", raising=False) + client = MagicMock() + assert await _resolve_app_project(client, "ApprovalApp") is None + + @pytest.mark.asyncio + async def test_resolve_app_action_schema(self): + client = self._client_returning( + { + "/FileOperations/Structure": { + "files": [{"name": "schema-1.json", "id": "f1"}], + "folders": [], + }, + "/FileOperations/File/f1": {"key": "schema-1", "inputs": []}, + } + ) + schema = await _resolve_app_action_schema(client, "proj-id") + assert schema is not None + assert schema["key"] == "schema-1" + + @pytest.mark.asyncio + async def test_resolve_app_action_schema_none_when_no_schema_file(self): + client = self._client_returning( + {"/FileOperations/Structure": {"files": [], "folders": []}} + ) + assert await _resolve_app_action_schema(client, "proj-id") is None + + +class TestEscalationJitFallbacks: + """Failure/fallback paths in the debug + JIT resolution flow.""" + + @pytest.fixture + def jit_resource(self): + 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", + app_type="Custom", + ), + recipients=[], + ) + ], + ) + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_PROJECT_ID": "proj-1"}, clear=False) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_debug_resolution_failure_falls_back_to_release( + self, + mock_interrupt, + mock_uipath_class, + mock_resolve_debug, + mock_resolve_project, + jit_resource, + ): + """If debug resolution raises, treat as release: skip JIT, still create.""" + os.environ.pop("UIPATH_FEATURE_EnableJITEscalationApps", None) + os.environ["UIPATH_FEATURE_EnableJITEscalationApps"] = "true" + try: + 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 + + mock_resolve_debug.side_effect = RuntimeError("job lookup failed") + + tool = create_escalation_tool(jit_resource) + call = ToolCall(args={}, id="test-call", name=tool.name) + await tool.awrapper(tool, call, {}) # type: ignore[attr-defined] + + # Debug unknown -> release: no JIT resolution, task still created. + mock_resolve_project.assert_not_called() + mock_client.tasks.create_async.assert_called_once() + finally: + os.environ.pop("UIPATH_FEATURE_EnableJITEscalationApps", None) + + @pytest.mark.asyncio + @patch.dict( + os.environ, + { + "UIPATH_PROJECT_ID": "proj-1", + "UIPATH_FEATURE_EnableJITEscalationApps": "true", + }, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_app_resolution_failure_raises_user_error( + self, + mock_interrupt, + mock_uipath_class, + mock_resolve_debug, + mock_resolve_project, + jit_resource, + ): + """If app resolution raises, the exception is logged and the missing-fields + USER error is raised (no task created).""" + 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 + + mock_resolve_debug.return_value = True + mock_resolve_project.side_effect = RuntimeError("studio backend 500") + + tool = create_escalation_tool(jit_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() + + +class TestResolveIsDebugRunEdgeCases: + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_JOB_KEY": "job-1"}) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + async def test_empty_parent_context_returns_false(self, mock_uipath_class): + job = MagicMock() + job.parent_context = None + mock_client = MagicMock() + mock_client.jobs.retrieve_async = AsyncMock(return_value=job) + mock_uipath_class.return_value = mock_client + assert await _resolve_is_debug_run() is False + + @pytest.mark.asyncio + @patch.dict(os.environ, {"UIPATH_JOB_KEY": "job-1"}) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + async def test_invalid_json_parent_context_returns_false(self, mock_uipath_class): + job = MagicMock() + job.parent_context = "not-valid-json" + mock_client = MagicMock() + mock_client.jobs.retrieve_async = AsyncMock(return_value=job) + mock_uipath_class.return_value = mock_client + assert await _resolve_is_debug_run() is False + + +class TestEscalationJitAppTypeDerivation: + @pytest.mark.asyncio + @patch.dict( + os.environ, + { + "UIPATH_PROJECT_ID": "proj-1", + "UIPATH_FEATURE_EnableJITEscalationApps": "true", + }, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_action_schema", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_app_project", + new_callable=AsyncMock, + ) + @patch( + "uipath_langchain.agent.tools.escalation_jit._resolve_is_debug_run", + new_callable=AsyncMock, + ) + @patch("uipath_langchain.agent.tools.escalation_jit.UiPath") + @patch("uipath_langchain.agent.tools.escalation_tool.UiPath") + @patch("uipath_langchain._utils.durable_interrupt.decorator.interrupt") + async def test_app_type_derived_from_project_type_when_absent( + self, + mock_interrupt, + mock_uipath_class, + mock_jit_uipath_class, + mock_resolve_debug, + mock_resolve_project, + mock_resolve_schema, + ): + """When the channel carries no app_type, it is derived from the resolved + project's projectType (Process -> Custom).""" + resource = 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=0, # inline + resource_key="test-key", + app_type=None, # not provided by the frontend + ), + recipients=[], + ) + ], + ) + + 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 + + mock_resolve_debug.return_value = True + mock_resolve_project.return_value = { + "designId": "d1", + "id": "i1", + "projectType": "Process", + } + mock_resolve_schema.return_value = {"key": "schema-1"} + + tool = create_escalation_tool(resource) + 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_type"] == "Custom" + assert kwargs["app_project_key"] == "d1"