feat: JIT escalation apps support

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

@@ -39,6 +39,10 @@ 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,
+    is_debug: bool = False,
+    solution_id: Optional[str] = None,
 ) -> RequestSpec:
     field_list = []
     outcome_list = []
@@ -94,7 +98,7 @@ def _create_spec(
                 )
 
     json_payload: Dict[str, Any] = {
-        "appId": app_key,
+        "appId": app_key if app_key is not None else f"ID{uuid.uuid4().hex}",
         "title": title,
         "data": data if data is not None else {},
         "actionableMessageMetaData": actionable_message_metadata
@@ -119,11 +123,16 @@ def _create_spec(
         ),
     }
 
+    if is_debug and app_type is not None:
+        json_payload["appType"] = app_type
+    if is_debug and app_project_key is not None:
+        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)
-
+    _apply_task_source(json_payload, source_name, is_debug, solution_id)
+    print('Calling CreateAppTask', json_payload)
     return RequestSpec(
         method="POST",
         endpoint=Endpoint("/orchestrator_/tasks/AppTasks/CreateAppTask"),
@@ -159,14 +168,20 @@ 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,
+    is_debug: Optional[bool] = None,
+    solution_id: Optional[str] = None,
+) -> 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 = solution_id or UiPathConfig.studio_solution_id
     if not (project_id and trace_id):
         return
     payload["taskSource"] = {
@@ -180,6 +195,12 @@ def _apply_task_source(payload: Dict[str, Any], source_name: str) -> None:
         },
     }
 
+    if solution_id is not None:
+        payload["taskSource"]["solutionId"] = solution_id
+    if is_debug:
+        payload["taskSource"]["isDebug"] = True
+        payload["taskSource"]["jobId"] = UiPathConfig.job_key
+
 
 def _normalize_priority(priority: str | None) -> str | None:
     """Normalize priority string to match API expectations.
@@ -229,6 +250,7 @@ def _create_quickform_spec(
     is_actionable_message_enabled: Optional[bool] = None,
     actionable_message_metadata: Optional[Dict[str, Any]] = None,
     source_name: str = "Agent",
+    solution_id: Optional[str] = None,
 ) -> RequestSpec:
     """Build the RequestSpec for Orchestrator's GenericTasks/CreateTask endpoint.
 
@@ -259,7 +281,7 @@ def _create_quickform_spec(
     )
     if actionable_message_metadata is not None:
         json_payload["actionableMessageMetaData"] = actionable_message_metadata
-    _apply_task_source(json_payload, source_name)
+    _apply_task_source(json_payload, source_name, solution_id=solution_id)
 
     return RequestSpec(
         method="POST",
@@ -459,6 +481,11 @@ 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,
+        is_debug: bool = False,
+        action_schema: Optional[TaskSchema] = None,
+        solution_id: Optional[str] = None,
     ) -> Task:
         """Creates a new action asynchronously.
 
@@ -478,34 +505,53 @@ 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. Only sent when is_debug is True.
+            app_type: Optional app type ("Custom" or "Coded"), forwarded for JIT (debug) task creation.
+            is_debug: When True, skips deployed-app key resolution and relies on app_project_key so a
+                not-yet-deployed app can be targeted during a debug run.
 
         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(
-                app_name, app_folder_path, app_folder_key
+
+        if is_debug and app_project_key:
+            (key, schema) = (app_key, TaskSchema(
+                    key=action_schema["key"],
+                    in_outs=action_schema["inOuts"],
+                    inputs=action_schema["inputs"],
+                    outputs=action_schema["outputs"],
+                    outcomes=action_schema["outcomes"],
+                ))
+        else:
+            (key, schema) = (
+                (app_key, None)
+                if app_key
+                else 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,
+            is_debug=is_debug,
+            solution_id=solution_id,
         )
-
+        print("Headers", spec.headers)
         response = await self.request_async(
             spec.method,
             spec.endpoint,
@@ -545,6 +591,11 @@ 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,
+        is_debug: bool = False,
+        action_schema: Optional[TaskSchema] = None,
+        solution_id: Optional[str] = None,
     ) -> Task:
         """Creates a new task synchronously.
 
@@ -564,30 +615,45 @@ def create(
             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. Only sent when is_debug is True.
+            app_type: Optional app type ("Custom" or "Coded"), forwarded for JIT (debug) task creation.
+            is_debug: When True, skips deployed-app key resolution and relies on app_project_key so a
+                not-yet-deployed app can be targeted during a debug run.
 
         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)
-        )
+
+        if is_debug:
+            (key, schema) = (app_key, action_schema)
+        else:
+            (key, schema) = (
+                (app_key, None)
+                if app_key
+                else 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,
+            is_debug=is_debug,
+            solution_id=solution_id,
         )
 
         response = self.request(
@@ -625,6 +691,7 @@ async def create_quickform_async(
         actionable_message_metadata: Optional[Dict[str, Any]] = None,
         creator_job_key: Optional[str] = None,
         source_name: str = "Agent",
+        solution_id: Optional[str] = None,
     ) -> Task:
         """Creates a new QuickForm task asynchronously.
 
@@ -675,6 +742,7 @@ async def create_quickform_async(
             is_actionable_message_enabled=is_actionable_message_enabled,
             actionable_message_metadata=actionable_message_metadata,
             source_name=source_name,
+            solution_id=solution_id,
         )
 
         response = await self.request_async(
@@ -715,6 +783,7 @@ def create_quickform(
         actionable_message_metadata: Optional[Dict[str, Any]] = None,
         creator_job_key: Optional[str] = None,
         source_name: str = "Agent",
+        solution_id: Optional[str] = None,
     ) -> Task:
         """Create a new QuickForm task synchronously.
 
@@ -733,6 +802,7 @@ def create_quickform(
             is_actionable_message_enabled=is_actionable_message_enabled,
             actionable_message_metadata=actionable_message_metadata,
             source_name=source_name,
+            solution_id=solution_id,
         )
 
         response = self.request(

packages/uipath-platform/tests/services/test_actions_service.py

@@ -128,6 +128,72 @@ def test_create_with_app_key(
         assert action.id == 1
         assert action.title == "Test Action"
 
+    def test_create_jit_debug_skips_resolution_and_sends_project_key(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+    ) -> None:
+        # JIT (debug): the app is not deployed, so no deployed-app schema lookup should happen and
+        # the project key + app type are sent so Orchestrator can resolve the app.
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/tasks/AppTasks/CreateAppTask",
+            status_code=200,
+            json={"id": 1, "title": "Test Action"},
+        )
+
+        action = service.create(
+            title="Test Action",
+            data={"test": "data"},
+            app_project_key="proj-key-abc",
+            app_type="Custom",
+            is_debug=True,
+        )
+
+        assert isinstance(action, Task)
+        requests = httpx_mock.get_requests()
+        # Only the create call — no deployed-action-apps-schemas resolution in debug mode.
+        assert all("deployed-action-apps-schemas" not in str(r.url) for r in requests)
+        create_request = [r for r in requests if "CreateAppTask" in str(r.url)][0]
+        body = json.loads(create_request.content)
+        assert body["appType"] == "Custom"
+        assert body["appProjectKey"] == "proj-key-abc"
+        assert body["appId"] is None
+
+    def test_create_does_not_send_project_key_when_not_debug(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+    ) -> None:
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/tasks/AppTasks/CreateAppTask",
+            status_code=200,
+            json={"id": 1, "title": "Test Action"},
+        )
+
+        action = service.create(
+            title="Test Action",
+            app_key="test-app-key",
+            data={"test": "data"},
+            app_project_key="proj-key-abc",
+            app_type="Coded",
+            is_debug=False,
+        )
+
+        assert isinstance(action, Task)
+        create_request = [
+            r for r in httpx_mock.get_requests() if "CreateAppTask" in str(r.url)
+        ][0]
+        body = json.loads(create_request.content)
+        # app type is still forwarded, but the project key is only sent in debug mode.
+        assert body["appType"] == "Coded"
+        assert "appProjectKey" not in body
+
     def test_create_with_assignee(
         self,
         httpx_mock: HTTPXMock,

packages/uipath/src/uipath/agent/models/agent.py

@@ -816,6 +816,14 @@ def _resolve_task_title(v: Any) -> Any:
 class BaseEscalationChannelProperties(BaseResourceProperties):
     """Fields shared by every escalation channel's properties."""
 
+    app_name: str | None = Field(default=None, alias="appName")
+    app_version: int = Field(..., alias="appVersion")
+    folder_name: Optional[str] = Field(None, alias="folderName")
+    resource_key: str | None = Field(default=None, alias="resourceKey")
+    project_key: str | None = Field(default=None, alias="projectKey")
+    app_type: str | None = Field(default=None, alias="appType")
+    solution_id: str | None = Field(default=None, alias="solutionId")
+    action_schema: Optional[Any] = Field(default=None, alias="actionSchema")
     is_actionable_message_enabled: Optional[bool] = Field(
         None, alias="isActionableMessageEnabled"
     )