fix: add support for multiple resume triggers

Author: cristipufu

src/uipath/runtime/resumable/protocols.py

@@ -8,8 +8,10 @@
 class UiPathResumableStorageProtocol(Protocol):
     """Protocol for storing and retrieving resume triggers."""
 
-    async def save_trigger(self, runtime_id: str, trigger: UiPathResumeTrigger) -> None:
-        """Save a resume trigger to storage.
+    async def save_triggers(
+        self, runtime_id: str, trigger: list[UiPathResumeTrigger]
+    ) -> None:
+        """Save resume triggers to storage.
 
         Args:
             trigger: The resume trigger to persist
@@ -19,17 +21,31 @@ async def save_trigger(self, runtime_id: str, trigger: UiPathResumeTrigger) -> N
         """
         ...
 
-    async def get_latest_trigger(self, runtime_id: str) -> UiPathResumeTrigger | None:
-        """Retrieve the most recent resume trigger from storage.
+    async def get_triggers(self, runtime_id: str) -> list[UiPathResumeTrigger] | None:
+        """Retrieve the resume triggers from storage.
 
         Returns:
-            The latest resume trigger, or None if no triggers exist
+            The resume triggers, or None if no triggers exist
 
         Raises:
             Exception: If retrieval operation fails
         """
         ...
 
+    async def delete_trigger(
+        self, runtime_id: str, trigger: UiPathResumeTrigger
+    ) -> None:
+        """Delete resume trigger from storage.
+
+        Args:
+            runtime_id: The runtime ID
+            trigger: The resume trigger to delete
+
+        Raises:
+            Exception: If deletion operation fails
+        """
+        ...
+
     async def set_value(
         self, runtime_id: str, namespace: str, key: str, value: Any
     ) -> None:

src/uipath/runtime/resumable/runtime.py

@@ -3,6 +3,8 @@
 import logging
 from typing import Any, AsyncGenerator
 
+from uipath.core.errors import UiPathPendingTriggerError
+
 from uipath.runtime.base import (
     UiPathExecuteOptions,
     UiPathRuntimeProtocol,
@@ -111,21 +113,29 @@ async def _restore_resume_input(
             input: User-provided input (takes precedence)
 
         Returns:
-            Input to use for resume, either provided or from storage
+            Input to use for resume: {interrupt_id: resume_data, ...}
         """
         # If user provided explicit input, use it
         if input is not None:
             return input
 
-        # Otherwise, fetch from storage
-        trigger = await self.storage.get_latest_trigger(self.runtime_id)
-        if not trigger:
+        # Fetch all triggers from storage
+        triggers = await self.storage.get_triggers(self.runtime_id)
+        if not triggers:
             return None
 
-        # Read trigger data via trigger_manager
-        resume_data = await self.trigger_manager.read_trigger(trigger)
+        # Build resume map: {interrupt_id: resume_data}
+        resume_map: dict[str, Any] = {}
+        for trigger in triggers:
+            try:
+                data = await self.trigger_manager.read_trigger(trigger)
+                resume_map[trigger.interrupt_id] = data
+                await self.storage.delete_trigger(self.runtime_id, trigger)
+            except UiPathPendingTriggerError:
+                # Trigger still pending, skip it
+                pass
 
-        return resume_data
+        return resume_map
 
     async def _handle_suspension(
         self, result: UiPathRuntimeResult
@@ -142,22 +152,31 @@ async def _handle_suspension(
         if isinstance(result, UiPathBreakpointResult):
             return result
 
-        # Check if trigger already exists in result
-        if result.trigger:
-            await self.storage.save_trigger(self.runtime_id, result.trigger)
-            return result
-
         suspended_result = UiPathRuntimeResult(
             status=UiPathRuntimeStatus.SUSPENDED,
             output=result.output,
         )
 
-        if result.output:
-            suspended_result.trigger = await self.trigger_manager.create_trigger(
-                result.output
+        # Get existing triggers and current interrupts
+        suspended_result.triggers = (
+            await self.storage.get_triggers(self.runtime_id) or []
+        )
+        current_interrupts = result.output or {}
+
+        # Diff: find new interrupts
+        existing_ids = {t.interrupt_id for t in suspended_result.triggers}
+        new_ids = current_interrupts.keys() - existing_ids
+
+        # Create triggers only for new interrupts
+        for interrupt_id in new_ids:
+            trigger = await self.trigger_manager.create_trigger(
+                current_interrupts[interrupt_id]
             )
+            trigger.interrupt_id = interrupt_id
+            suspended_result.triggers.append(trigger)
 
-            await self.storage.save_trigger(self.runtime_id, suspended_result.trigger)
+        if suspended_result.triggers:
+            await self.storage.save_triggers(self.runtime_id, suspended_result.triggers)
 
         return suspended_result
 

src/uipath/runtime/resumable/trigger.py

@@ -49,6 +49,7 @@ class UiPathApiTrigger(BaseModel):
 class UiPathResumeTrigger(BaseModel):
     """Information needed to resume execution."""
 
+    interrupt_id: str = Field(alias="interruptId")
     trigger_type: UiPathResumeTriggerType = Field(
         default=UiPathResumeTriggerType.API, alias="triggerType"
     )