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

[Chibionos]

Summary

Implements the suspend/resume pattern for RPA process invocations within the evaluation runtime. When a LangGraph agent calls interrupt(InvokeProcess(...)), the runtime now correctly:

• ✅ Detects the interrupt and extracts trigger data
• ✅ Returns SUSPENDED status instead of SUCCESSFUL
• ✅ Skips evaluator execution until resumed
• ✅ Preserves trigger information for the serverless executor

This enables evaluations to pause while waiting for external RPA jobs to complete, then resume with the job results.

---

Architecture

Flow Overview

```
Agent Execution → LangGraph interrupt() → Functions Runtime Detection
↓
Create UiPathResumeTrigger
↓
Return SUSPENDED status
↓
Evaluation Runtime (skip evaluators)
↓
Publish suspended event to bus
↓
Serverless Executor (poll job)
↓
Resume with job results
```

Key Components

1. Functions Runtime (src/uipath/functions/runtime.py)

• New method: _detect_langgraph_interrupt()
  • Detects __interrupt__ field in agent output
  • Extracts InvokeProcess object from interrupt
  • Converts to UiPathResumeTrigger with job metadata
• Modified: execute() to check interrupts and return SUSPENDED status

2. Evaluation Runtime (src/uipath/_cli/_evals/_runtime.py)

• Lines 470-527: SUSPENDED status handling
  • Detects when agent returns SUSPENDED status
  • Populates agentExecutionOutput with trigger data
  • Skips evaluator execution (preserves for resume)
  • Publishes EvalRunUpdatedEvent with suspended state

3. Test Agent (samples/event-trigger/test_suspend_resume_agent.py)

• Uses MemorySaver checkpointer (required for interrupts)
• Returns raw dict to preserve __interrupt__ field
• Demonstrates proper interrupt pattern for RPA invocation

---

Changes Made

Files Modified

File	Lines Changed	Description
src/uipath/functions/runtime.py	+70	Added interrupt detection and trigger extraction
src/uipath/_cli/_evals/_runtime.py	+58	Added SUSPENDED status handling in evaluation flow
samples/event-trigger/test_suspend_resume_agent.py	+126	Created test agent demonstrating the pattern

Detailed Changes

Functions Runtime

# New imports (lines 26-30)
from uipath.runtime.resumable.trigger import (
    UiPathResumeTrigger,
    UiPathResumeTriggerType,
    UiPathResumeTriggerName,
)

# New method (lines 132-195)
def _detect_langgraph_interrupt(
    self, output: dict[str, Any]
) -> UiPathResumeTrigger | None:
    """Detect LangGraph __interrupt__ field and extract InvokeProcess trigger."""
    # Checks for __interrupt__ array in output
    # Extracts InvokeProcess from interrupt.value
    # Creates UiPathResumeTrigger with job metadata

# Modified execute() (lines 207-217)
- Calls _detect_langgraph_interrupt() on output
- Returns SUSPENDED status with trigger when interrupt detected

Evaluation Runtime

# Lines 470-527: Suspend detection and handling
if agent_execution_output.result.status == UiPathRuntimeStatus.SUSPENDED:
    # Extract triggers from result
    # Populate agentExecutionOutput (important for serverless executor)
    # Publish EvalRunUpdatedEvent with suspended status
    # Return early - evaluators run on resume

---

Testing Instructions

Prerequisites

# Ensure you're in the event-trigger sample directory
cd samples/event-trigger

# Activate the virtual environment
source .venv/bin/activate

# Verify the package is installed in editable mode
pip show uipath-python
# Should show: Location: /home/chibionos/r2/uipath-python/src

Test 1: Direct Runtime Execution

Test the functions runtime directly to verify interrupt detection:

# Run the agent directly
uv run python -c "
import asyncio
from uipath.functions.runtime import UiPathFunctionsRuntime

async def test():
    runtime = UiPathFunctionsRuntime(
        'test_suspend_resume_agent.py',
        'main',
        'test_suspend_resume_agent'
    )
    result = await runtime.execute({'query': 'Test suspend and resume'})
    print(f'Status: {result.status}')
    print(f'Trigger: {result.trigger}')
    if result.trigger:
        print(f'Process: {result.trigger.payload[\"process_name\"]}')
        print(f'Arguments: {result.trigger.payload[\"input_arguments\"]}')

asyncio.run(test())
"

Expected Output:

Status: UiPathRuntimeStatus.SUSPENDED
Trigger: UiPathResumeTrigger(...)
Process: TestProcess
Arguments: {'query': 'Test suspend and resume', 'timestamp': '...'}

Test 2: Evaluation Runtime

Test the complete evaluation flow:

# Run evaluation (creates evaluation JSON)
uv run uipath eval test_suspend_resume_agent evaluations/test_suspend_resume.json --output-file test_output.json

# Check the output for suspended status
cat test_output.json | python -m json.tool

Expected JSON Structure:

{
  "evaluationSetName": "Suspend Resume Test",
  "evaluationSetResults": [{
    "evaluationName": "Basic suspend test",
    "evaluationRunResults": [],
    "agentExecutionOutput": {
      "result": {
        "status": "suspended",
        "trigger": {
          "trigger_type": "JOB",
          "item_key": "job-<uuid>",
          "folder_path": "Shared",
          "payload": {
            "process_name": "TestProcess",
            "input_arguments": {...}
          }
        }
      }
    }
  }]
}

Test 3: Check Logs

Verify proper logging throughout the flow:

# Run with verbose logging
uv run uipath eval test_suspend_resume_agent evaluations/test_suspend_resume.json 2>&1 | tee test.log

# Check for key log messages
grep "Detected LangGraph interrupt" test.log
grep "Evaluation execution suspended" test.log
grep "SUSPENDED" test.log

Expected Log Messages:

• Detected LangGraph interrupt - suspending execution for process: TestProcess
• Evaluation execution suspended for eval 'Basic suspend test'
• Status showing SUSPENDED

---

Verification Checklist

Before approving this PR, verify:

• Functions runtime correctly detects __interrupt__ field
• Runtime returns SUSPENDED status with proper trigger
• Trigger contains process name and input arguments
• Evaluation runtime skips evaluators when suspended
• agentExecutionOutput is populated in JSON output
• Suspended status is published to event bus
• Test agent uses checkpointer and returns raw dict
• All existing unit tests pass
• Direct runtime execution test succeeds
• Evaluation test succeeds with correct JSON output

---

Integration with Serverless Executor

The serverless executor (external component) will:

1. Detect SUSPENDED status from agentExecutionOutput
2. Extract trigger from evaluation output JSON
3. Start polling the RPA job using trigger metadata
4. Wait for job completion
5. Resume evaluation with job results
6. Run evaluators on final output

This PR provides the agent-side infrastructure. The serverless executor changes are separate.

---

Related Work

• Pattern follows UiPathResumableRuntime implementation
• Compatible with existing trigger infrastructure
• Maintains backward compatibility (no changes to non-suspended flows)
• Test agent demonstrates best practices for interrupt usage

---

Technical Notes

Why Check for __interrupt__ Field?

LangGraph's interrupt() function only creates the __interrupt__ field when:

1. A checkpointer is configured (e.g., MemorySaver)
2. The graph is compiled with the checkpointer
3. The function returns the raw dict (not wrapped in Pydantic)

Our runtime must detect this field to distinguish suspended executions from normal completions.

Why Skip Evaluators?

When an evaluation suspends:

• The agent hasn't produced final output yet (it's waiting for RPA job)
• Evaluators would fail or produce meaningless results
• We need to preserve the evaluation state for resumption
• Evaluators run after resume when we have the final output

Why Populate agentExecutionOutput?

The serverless executor needs to:

• Detect that execution suspended (from status)
• Extract trigger data to start polling
• Know what job to wait for
• Have all metadata to resume correctly

Without agentExecutionOutput, the executor wouldn't know how to resume the evaluation.

---

Next Steps

After merging this PR:

1. Update serverless executor to handle SUSPENDED status
2. Implement job polling and resume logic
3. Add integration tests for end-to-end flow
4. Update documentation for interrupt pattern
5. Consider adding more sample agents using this pattern

---

Dev Tag

This PR introduces new functionality that needs testing before release. Tag: v2.5.0-dev.suspend-resume

[Chibionos]

Quick Testing Guide

Install the Dev Version

pip install git+https://github.com/UiPath/uipath-python@v2.4.9-dev.suspend-resume

Run Quick Test

cd samples/event-trigger

# Test 1: Direct runtime test
python -c "
import asyncio
from uipath.functions.runtime import UiPathFunctionsRuntime

async def test():
    runtime = UiPathFunctionsRuntime(
        'test_suspend_resume_agent.py',
        'main', 
        'test_suspend_resume_agent'
    )
    result = await runtime.execute({'query': 'Test'})
    print(f'✓ Status: {result.status}')
    print(f'✓ Has Trigger: {result.trigger is not None}')
    if result.trigger:
        print(f'✓ Process: {result.trigger.payload[\"process_name\"]}')

asyncio.run(test())
"

# Test 2: Evaluation test
uipath eval test_suspend_resume_agent evaluations/test_suspend_resume.json --output-file test_output.json
cat test_output.json | python -m json.tool | grep -A 5 "status"

Expected Results

✅ Status: SUSPENDED
✅ Trigger exists with process name and arguments
✅ Evaluation output shows suspended state in JSON

Key Files Changed

• Functions Runtime: src/uipath/functions/runtime.py (+70 lines)

  • New _detect_langgraph_interrupt() method
  • Trigger extraction from __interrupt__ field

• Evaluation Runtime: src/uipath/_cli/_evals/_runtime.py (+58 lines)

  • SUSPENDED status detection
  • Skip evaluators when suspended
  • Publish suspended event

• Test Agent: samples/event-trigger/test_suspend_resume_agent.py (new file)

  • Demonstrates correct interrupt pattern
  • Uses checkpointer and returns raw dict

---

Tag: v2.4.9-dev.suspend-resume
Branch: investigate-rpa-samples