feat: just-in-time debug escalation apps support

Author: Sandeepan-Ghosh-0312

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.2.13"
+version = "0.2.14"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/action_center/_tasks_service.py

@@ -17,7 +17,7 @@
 from ..common._folder_context import FolderContext, header_folder
 from ..common._models import Endpoint, RequestSpec
 from .task_schema import TaskSchema
-from .tasks import Task, TaskRecipient, TaskRecipientType
+from .tasks import Task, TaskRecipient, TaskRecipientType, is_low_code_app
 
 
 def _ensure_string_value(value: Any) -> str:
@@ -39,6 +39,8 @@ def _create_spec(
     is_actionable_message_enabled: Optional[bool] = None,
     actionable_message_metadata: Optional[Dict[str, Any]] = None,
     source_name: str = "Agent",
+    app_project_key: Optional[str] = None,
+    app_type: Optional[str] = None,
 ) -> RequestSpec:
     field_list = []
     outcome_list = []
@@ -94,7 +96,7 @@ def _create_spec(
                 )
 
     json_payload: Dict[str, Any] = {
-        "appId": app_key,
+        "appId": app_key if app_key is not None else f"{uuid.UUID(int=0).hex}",
         "title": title,
         "data": data if data is not None else {},
         "actionableMessageMetaData": actionable_message_metadata
@@ -119,11 +121,14 @@ def _create_spec(
         ),
     }
 
+    if app_project_key is not None:
+        json_payload["appType"] = app_type
+        json_payload["appProjectKey"] = app_project_key
+
     _apply_priority_labels_and_actionable_toggle(
         json_payload, priority, labels, is_actionable_message_enabled
     )
     _apply_task_source(json_payload, source_name)
-
     return RequestSpec(
         method="POST",
         endpoint=Endpoint("/orchestrator_/tasks/AppTasks/CreateAppTask"),
@@ -159,14 +164,18 @@ def _apply_priority_labels_and_actionable_toggle(
         payload["isActionableMessageEnabled"] = is_actionable_message_enabled
 
 
-def _apply_task_source(payload: Dict[str, Any], source_name: str) -> None:
+def _apply_task_source(
+    payload: Dict[str, Any],
+    source_name: str,
+) -> None:
     """Populate ``payload["taskSource"]`` when UiPathConfig has project_id + trace_id.
 
     Shared between AppTask and QuickForm spec builders — the taskSource block is
     identical for both task types.
     """
     project_id = UiPathConfig.project_id
     trace_id = UiPathConfig.trace_id
+    solution_id = UiPathConfig.studio_solution_id
     if not (project_id and trace_id):
         return
     payload["taskSource"] = {
@@ -178,8 +187,12 @@ def _apply_task_source(payload: Dict[str, Any], source_name: str) -> None:
             "JobKey": UiPathConfig.job_key,
             "ProcessKey": UiPathConfig.process_uuid,
         },
+        "jobId": UiPathConfig.job_key
     }
 
+    if UiPathConfig.is_rooted_to_debug_job:
+        payload["taskSource"]["isDebug"] = True
+        payload["taskSource"]["solutionId"] = solution_id
 
 def _normalize_priority(priority: str | None) -> str | None:
     """Normalize priority string to match API expectations.
@@ -459,6 +472,9 @@ async def create_async(
         is_actionable_message_enabled: Optional[bool] = None,
         actionable_message_metadata: Optional[Dict[str, Any]] = None,
         source_name: str = "Agent",
+        app_project_key: Optional[str] = None,
+        app_type: Optional[str] = None,
+        action_schema: Optional[Dict[str, Any]] = None,
     ) -> Task:
         """Creates a new action asynchronously.
 
@@ -478,34 +494,48 @@ async def create_async(
             is_actionable_message_enabled: Optional boolean indicating whether actionable notifications are enabled for this task
             actionable_message_metadata: Optional metadata for the action
             source_name: The name of the source that created the task. Defaults to 'Agent'.
+            app_project_key: Optional project key of the app. Used for JIT (debug) task creation so
+                Orchestrator can resolve a not-yet-deployed app.
+            app_type: Optional app type ("Custom" or "Coded"), forwarded for JIT (debug) task creation.
 
         Returns:
             Action: The created action object
 
         Raises:
             Exception: If neither app_name nor app_key is provided for app-specific actions
         """
-        (key, action_schema) = (
-            (app_key, None)
-            if app_key
-            else await self._get_app_key_and_schema_async(
+        key: Optional[str]
+        schema: Optional[TaskSchema]
+        if app_project_key and is_low_code_app(app_type) and action_schema is not None:
+            key = app_key
+            schema = TaskSchema(
+                key=action_schema["key"],
+                in_outs=action_schema["inOuts"],
+                inputs=action_schema["inputs"],
+                outputs=action_schema["outputs"],
+                outcomes=action_schema["outcomes"],
+            )
+        elif app_key:
+            key, schema = app_key, None
+        else:
+            key, schema = await self._get_app_key_and_schema_async(
                 app_name, app_folder_path, app_folder_key
             )
-        )
         spec = _create_spec(
             title=title,
             data=data,
             app_key=key,
-            action_schema=action_schema,
+            action_schema=schema,
             app_folder_key=app_folder_key,
             app_folder_path=app_folder_path,
             priority=priority,
             labels=labels,
             is_actionable_message_enabled=is_actionable_message_enabled,
             actionable_message_metadata=actionable_message_metadata,
             source_name=source_name,
+            app_project_key=app_project_key,
+            app_type=app_type,
         )
-
         response = await self.request_async(
             spec.method,
             spec.endpoint,
@@ -545,49 +575,51 @@ def create(
         is_actionable_message_enabled: Optional[bool] = None,
         actionable_message_metadata: Optional[Dict[str, Any]] = None,
         source_name: str = "Agent",
+        app_project_key: Optional[str] = None,
+        app_type: Optional[str] = None,
+        action_schema: Optional[Dict[str, Any]] = None,
     ) -> Task:
         """Creates a new task synchronously.
 
-        This method creates a new action task in UiPath Orchestrator. The action can be
-        either app-specific (using app_name or app_key) or a generic action.
-
-        Args:
-            title: The title of the action
-            data: Optional dictionary containing input data for the action
-            app_name: The name of the application (if creating an app-specific action)
-            app_key: The key of the application (if creating an app-specific action)
-            app_folder_path: Optional folder path for the action
-            app_folder_key: Optional folder key for the action
-            assignee: Optional username or email to assign the task to
-            priority: Optional priority of the task
-            labels: Optional list of labels for the task
-            is_actionable_message_enabled: Optional boolean indicating  whether actionable notifications are enabled for this task
-            actionable_message_metadata: Optional metadata for the action
-            source_name: The name of the source that created the task. Defaults to 'Agent'.
+        See :meth:`create_async` for parameter docs.
 
         Returns:
             Action: The created action object
 
         Raises:
             Exception: If neither app_name nor app_key is provided for app-specific actions
         """
-        (key, action_schema) = (
-            (app_key, None)
-            if app_key
-            else self._get_app_key_and_schema(app_name, app_folder_path, app_folder_key)
-        )
+        key: Optional[str]
+        schema: Optional[TaskSchema]
+        if app_project_key and is_low_code_app(app_type) and action_schema is not None:
+            key = app_key
+            schema = TaskSchema(
+                key=action_schema["key"],
+                in_outs=action_schema["inOuts"],
+                inputs=action_schema["inputs"],
+                outputs=action_schema["outputs"],
+                outcomes=action_schema["outcomes"],
+            )
+        elif app_key:
+            key, schema = app_key, None
+        else:
+            key, schema = self._get_app_key_and_schema(
+                app_name, app_folder_path, app_folder_key
+            )
         spec = _create_spec(
             title=title,
             data=data,
             app_key=key,
-            action_schema=action_schema,
+            action_schema=schema,
             app_folder_key=app_folder_key,
             app_folder_path=app_folder_path,
             priority=priority,
             labels=labels,
             is_actionable_message_enabled=is_actionable_message_enabled,
             actionable_message_metadata=actionable_message_metadata,
             source_name=source_name,
+            app_project_key=app_project_key,
+            app_type=app_type,
         )
 
         response = self.request(

packages/uipath-platform/src/uipath/platform/action_center/tasks.py

@@ -6,6 +6,12 @@
 
 from pydantic import BaseModel, ConfigDict, Field, field_serializer
 
+APP_TYPE_LOW_CODE = "Custom"
+
+def is_low_code_app(app_type: str | None) -> bool:
+    """Return True when ``app_type`` denotes a low-code (custom) app."""
+    return app_type == APP_TYPE_LOW_CODE
+
 
 class TaskStatus(enum.IntEnum):
     """Enum representing possible Task status."""

packages/uipath-platform/src/uipath/platform/common/_config.py

@@ -14,6 +14,7 @@ class UiPathApiConfig(BaseModel):
 class ConfigurationManager:
     _instance = None
     studio_solution_id: str | None = None
+    _is_debug_run_override: bool | None = None
 
     def __new__(cls):
         if cls._instance is None:
@@ -194,8 +195,15 @@ def licensing_context(self) -> str | None:
     @property
     def is_rooted_to_debug_job(self) -> bool:
         """Whether this job, which may be a deployed process, is rooted to a debug session (e.g. Maestro solution debug)."""
+        if self._is_debug_run_override is not None:
+            return self._is_debug_run_override
         return self._read_internal_argument("isDebug") is True
 
+    @is_rooted_to_debug_job.setter
+    def is_rooted_to_debug_job(self, value: bool) -> None:
+        """Override the debug-run status resolved at runtime."""
+        self._is_debug_run_override = value
+
     @property
     def is_tracing_enabled(self) -> bool:
         from uipath.platform.constants import ENV_TRACING_ENABLED
@@ -205,6 +213,7 @@ def is_tracing_enabled(self) -> bool:
     def reset(self) -> None:
         """Reset mutable cached state to defaults."""
         self.studio_solution_id = None
+        self._is_debug_run_override = None
         # Invalidate cached_property by removing from instance __dict__
         self.__dict__.pop("_internal_arguments", None)
 

packages/uipath-platform/src/uipath/platform/orchestrator/job.py

@@ -80,4 +80,5 @@ class Job(BaseModel):
     has_warnings: Optional[bool] = Field(default=None, alias="HasWarnings")
     job_error: Optional[JobErrorInfo] = Field(default=None, alias="JobError")
     folder_key: Optional[str] = Field(default=None, alias="FolderKey")
+    parent_context: Optional[str] = Field(default=None, alias="ParentContext")
     id: int = Field(alias="Id")

packages/uipath-platform/tests/common/test_config_env_vars.py

@@ -190,3 +190,39 @@ def test_has_eval_folder(self, monkeypatch, tmp_path):
         assert UiPathConfig.has_eval_folder is False
         (tmp_path / EVALS_FOLDER).mkdir()
         assert UiPathConfig.has_eval_folder is True
+
+
+class TestIsRootedToDebugJob:
+    @pytest.fixture(autouse=True)
+    def _reset_singleton(self):
+        # is_rooted_to_debug_job mutates the process-wide singleton; reset around
+        # each test so nothing leaks between tests.
+        UiPathConfig.reset()
+        yield
+        UiPathConfig.reset()
+
+    def test_defaults_to_false_when_unset(self):
+        assert UiPathConfig.is_rooted_to_debug_job is False
+
+    def test_reads_is_debug_internal_argument(self):
+        UiPathConfig.__dict__["_internal_arguments"] = {"isDebug": True}
+        assert UiPathConfig.is_rooted_to_debug_job is True
+
+    def test_setter_override_wins_and_is_true(self):
+        UiPathConfig.is_rooted_to_debug_job = True
+        assert UiPathConfig.is_rooted_to_debug_job is True
+
+    def test_setter_override_beats_internal_argument(self):
+        # Even when the internal argument says True, an explicit False override wins.
+        UiPathConfig.__dict__["_internal_arguments"] = {"isDebug": True}
+        UiPathConfig.is_rooted_to_debug_job = False
+        assert UiPathConfig.is_rooted_to_debug_job is False
+
+    def test_reset_clears_override(self):
+        UiPathConfig.__dict__["_internal_arguments"] = {"isDebug": True}
+        UiPathConfig.is_rooted_to_debug_job = False
+        assert UiPathConfig.is_rooted_to_debug_job is False
+        UiPathConfig.reset()
+        # Override cleared, so it falls back to the internal argument again — but
+        # reset() also drops the cached internal arguments, so it re-reads (None).
+        assert UiPathConfig.is_rooted_to_debug_job is False