feat: Add suspend/resume support for RPA invocations in evaluations

This change implements the suspend/resume pattern for RPA process invocations
within the evaluation runtime, ensuring evaluations can pause when waiting for
external job completion and resume after the job finishes.

Key Changes:

1. Functions Runtime (src/uipath/functions/runtime.py):
   - Added _detect_langgraph_interrupt() method to detect LangGraph's
     __interrupt__ field and extract trigger information
   - Modified execute() to check for interrupts and return SUSPENDED status
   - Converts InvokeProcess objects to UiPathResumeTrigger for serverless executor

2. Evaluation Runtime (src/uipath/_cli/_evals/_runtime.py):
   - Added SUSPENDED status detection after agent execution
   - Populates agentExecutionOutput with trigger data when suspended
   - Skips evaluator execution for suspended runs
   - Publishes EvalRunUpdatedEvent with suspended status to event bus

3. Test Agent (samples/event-trigger/test_suspend_resume_agent.py):
   - Updated to use MemorySaver checkpointer (required for LangGraph interrupts)
   - Returns raw dict to preserve __interrupt__ field for runtime detection

The implementation follows the established pattern from UiPathResumableRuntime
and ensures proper trigger extraction and status handling throughout the
evaluation lifecycle.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: Chibi Vikram

samples/event-trigger/test_suspend_resume_agent.py

@@ -0,0 +1,109 @@
+"""Test agent that demonstrates suspend/resume pattern with RPA process invocation."""
+
+from typing import TypedDict
+
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
+from pydantic import BaseModel
+
+from uipath.platform.common import InvokeProcess
+
+
+class Input(BaseModel):
+    """Input for the test agent."""
+
+    query: str
+
+
+class Output(BaseModel):
+    """Output from the test agent."""
+
+    result: str
+    process_output: dict | None = None
+
+
+class State(TypedDict):
+    """Agent state."""
+
+    query: str
+    process_result: dict | None
+    final_result: str
+
+
+def prepare_input(state: State) -> State:
+    """Prepare input for RPA process."""
+    print(f"Preparing to call RPA process with query: {state['query']}")
+    return state
+
+
+def call_rpa_process(state: State) -> State:
+    """Call RPA process - this will suspend execution."""
+    print("Calling RPA process - execution will suspend here")
+
+    # This interrupt() call will cause the runtime to suspend
+    # The serverless executor will detect SUSPENDED status, poll the job,
+    # and then resume execution once the job completes
+    process_result = interrupt(
+        InvokeProcess(
+            name="TestProcess",  # Replace with actual process name
+            input_arguments={"query": state["query"], "timestamp": "2024-01-08"},
+            process_folder_path="Shared",  # Replace with actual folder
+            process_folder_key=None,
+        )
+    )
+
+    print(f"RPA process completed with result: {process_result}")
+
+    return {**state, "process_result": process_result}
+
+
+def format_output(state: State) -> State:
+    """Format final output after RPA process completes."""
+    process_result = state.get("process_result", {})
+
+    final_result = (
+        f"Processed query '{state['query']}' via RPA. Result: {process_result}"
+    )
+
+    return {**state, "final_result": final_result}
+
+
+# Build the graph
+builder = StateGraph(State)
+builder.add_node("prepare", prepare_input)
+builder.add_node("call_rpa", call_rpa_process)
+builder.add_node("format", format_output)
+
+builder.add_edge(START, "prepare")
+builder.add_edge("prepare", "call_rpa")
+builder.add_edge("call_rpa", "format")
+builder.add_edge("format", END)
+
+
+def main(input_data: Input):
+    """Main entry point for the agent.
+
+    Returns raw dict to preserve __interrupt__ field for suspend/resume.
+    When execution suspends, the dict will contain __interrupt__ field with trigger data.
+    When execution completes, the dict will contain final_result.
+    """
+    from langgraph.checkpoint.memory import MemorySaver
+
+    # IMPORTANT: Must use checkpointer for interrupt() to work
+    checkpointer = MemorySaver()
+    graph = builder.compile(checkpointer=checkpointer)
+
+    # Generate unique thread ID for this execution
+    import uuid
+
+    thread_id = f"agent-{uuid.uuid4()}"
+
+    config = {"configurable": {"thread_id": thread_id}}
+
+    result = graph.invoke(
+        {"query": input_data.query, "process_result": None, "final_result": ""}, config
+    )
+
+    # Return raw dict - preserves __interrupt__ field if suspended
+    # Runtime will detect __interrupt__ and create UiPath trigger
+    return result

src/uipath/_cli/_evals/_runtime.py

@@ -432,6 +432,14 @@ async def _execute_eval(
                     agent_execution_output = await self.execute_runtime(
                         eval_item, execution_id, runtime
                     )
+
+                    logger.info(
+                        f"DEBUG: Agent execution result status: {agent_execution_output.result.status}"
+                    )
+                    logger.info(
+                        f"DEBUG: Agent execution result trigger: {agent_execution_output.result.trigger}"
+                    )
+
                 except Exception as e:
                     if self.context.verbose:
                         if isinstance(e, EvaluationRuntimeException):
@@ -467,6 +475,58 @@ async def _execute_eval(
                         )
                     raise
 
+                # Check if execution was suspended (e.g., waiting for RPA job completion)
+                if (
+                    agent_execution_output.result.status
+                    == UiPathRuntimeStatus.SUSPENDED
+                ):
+                    # For suspended executions, we don't run evaluators yet
+                    # The serverless executor should save the triggers and resume later
+                    logger.info(
+                        f"Evaluation execution suspended for eval '{eval_item.name}' (id: {eval_item.id})"
+                    )
+
+                    # Extract triggers from result
+                    triggers = []
+                    if agent_execution_output.result.trigger:
+                        triggers.append(agent_execution_output.result.trigger)
+                    if agent_execution_output.result.triggers:
+                        triggers.extend(agent_execution_output.result.triggers)
+
+                    # IMPORTANT: Always include execution output with triggers when suspended
+                    # This ensures triggers are visible in the output JSON for serverless executor
+                    evaluation_run_results.agent_execution_output = (
+                        convert_eval_execution_output_to_serializable(
+                            agent_execution_output
+                        )
+                    )
+
+                    # Publish suspended status event
+                    await self.event_bus.publish(
+                        EvaluationEvents.UPDATE_EVAL_RUN,
+                        EvalRunUpdatedEvent(
+                            execution_id=execution_id,
+                            eval_item=eval_item,
+                            eval_results=[],
+                            success=True,  # Not failed, just suspended
+                            agent_output={
+                                "status": "suspended",
+                                "triggers": [
+                                    t.model_dump(by_alias=True) for t in triggers
+                                ],
+                            },
+                            agent_execution_time=agent_execution_output.execution_time,
+                            spans=agent_execution_output.spans,
+                            logs=agent_execution_output.logs,
+                            exception_details=None,
+                        ),
+                        wait_for_completion=False,
+                    )
+
+                    # Return partial results with trigger information
+                    # The evaluation will be completed when resumed
+                    return evaluation_run_results
+
                 if self.context.verbose:
                     evaluation_run_results.agent_execution_output = (
                         convert_eval_execution_output_to_serializable(

src/uipath/functions/runtime.py

@@ -23,6 +23,11 @@
     UiPathRuntimeError,
 )
 from uipath.runtime.schema import UiPathRuntimeSchema
+from uipath.runtime.resumable.trigger import (
+    UiPathResumeTrigger,
+    UiPathResumeTriggerType,
+    UiPathResumeTriggerName,
+)
 
 from .schema_gen import get_type_schema
 from .type_conversion import (
@@ -124,6 +129,71 @@ async def _execute_function(
 
         return convert_from_class(result) if result is not None else {}
 
+    def _detect_langgraph_interrupt(
+        self, output: dict[str, Any]
+    ) -> UiPathResumeTrigger | None:
+        """Detect LangGraph __interrupt__ field and extract InvokeProcess trigger.
+
+        LangGraph's interrupt() creates an __interrupt__ field in the output dict:
+        {
+            "query": "...",
+            "final_result": "",
+            "__interrupt__": [Interrupt(value=InvokeProcess(...), id="...")]
+        }
+
+        We extract the InvokeProcess from the interrupt and convert it to a UiPath trigger.
+        """
+        try:
+            if not isinstance(output, dict):
+                return None
+
+            # Check for LangGraph's __interrupt__ field
+            if "__interrupt__" not in output:
+                return None
+
+            interrupts = output["__interrupt__"]
+            if not interrupts or not isinstance(interrupts, list):
+                logger.warning("__interrupt__ field exists but is not a list")
+                return None
+
+            # Extract first interrupt
+            interrupt_obj = interrupts[0]
+            if not hasattr(interrupt_obj, "value"):
+                logger.warning("Interrupt object missing 'value' attribute")
+                return None
+
+            invoke_process = interrupt_obj.value
+
+            # Check if it's an InvokeProcess object (has name and input_arguments)
+            if not (
+                hasattr(invoke_process, "name")
+                and hasattr(invoke_process, "input_arguments")
+            ):
+                logger.warning(
+                    f"Interrupt value is not InvokeProcess (type: {type(invoke_process)})"
+                )
+                return None
+
+            logger.info(
+                f"Detected LangGraph interrupt - suspending execution for process: {invoke_process.name}"
+            )
+
+            # Convert InvokeProcess to UiPath trigger
+            return UiPathResumeTrigger(
+                trigger_type=UiPathResumeTriggerType.JOB,
+                trigger_name=UiPathResumeTriggerName.JOB,
+                item_key=f"job-{uuid.uuid4()}",  # Generate unique job key
+                folder_path=getattr(invoke_process, "process_folder_path", "Shared"),
+                payload={
+                    "process_name": invoke_process.name,
+                    "input_arguments": invoke_process.input_arguments or {},
+                    "folder_key": getattr(invoke_process, "process_folder_key", None),
+                },
+            )
+        except Exception as e:
+            logger.warning(f"Failed to detect LangGraph interrupt: {e}")
+            return None
+
     async def execute(
         self,
         input: dict[str, Any] | None = None,
@@ -134,6 +204,21 @@ async def execute(
             func = self._load_function()
             output = await self._execute_function(func, input or {})
 
+            logger.info(f"Output type: {type(output)}, has __interrupt__: {'__interrupt__' in output if isinstance(output, dict) else False}")
+
+            # Check if output represents a LangGraph interrupt (suspend)
+            trigger = self._detect_langgraph_interrupt(output)
+            logger.info(f"Trigger detected: {trigger}")
+            if trigger:
+                logger.info(
+                    f"Detected LangGraph interrupt - suspending execution with trigger: {trigger.item_key}"
+                )
+                return UiPathRuntimeResult(
+                    output=None,  # No final output yet (suspended)
+                    status=UiPathRuntimeStatus.SUSPENDED,
+                    trigger=trigger,
+                )
+
             return UiPathRuntimeResult(
                 output=output,
                 status=UiPathRuntimeStatus.SUCCESSFUL,