feat: support multiple resume triggers per interrupt

Author: radu-mocanu

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.11.5"
+version = "0.12.0"
 description = "Runtime abstractions and interfaces for building agents and automation scripts in the UiPath ecosystem"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath/runtime/resumable/protocols.py

@@ -68,6 +68,23 @@ async def create_trigger(self, suspend_value: Any) -> UiPathResumeTrigger:
         """
         ...
 
+    async def create_triggers(self, suspend_value: Any) -> list[UiPathResumeTrigger]:
+        """Create resume triggers from a suspend value.
+
+        Most suspend values produce one trigger. Composite values may produce
+        multiple sibling triggers for the same interrupt.
+
+        Args:
+            suspend_value: The value that caused the suspension.
+
+        Returns:
+            UiPathResumeTrigger objects ready to be persisted.
+
+        Raises:
+            UiPathRuntimeError: If trigger creation fails
+        """
+        ...
+
 
 class UiPathResumeTriggerReaderProtocol(Protocol):
     """Protocol for reading resume triggers and converting them to runtime input."""

src/uipath/runtime/resumable/runtime.py

@@ -1,6 +1,8 @@
 """Resumable runtime protocol and implementation."""
 
 import logging
+from collections import Counter
+from enum import Enum
 from typing import Any, AsyncGenerator
 
 from uipath.core.errors import UiPathPendingTriggerError
@@ -25,6 +27,8 @@
 
 logger = logging.getLogger(__name__)
 
+UIPATH_METADATA_KEY = "__uipath"
+
 
 class UiPathResumableRuntime:
     """Generic runtime wrapper that adds resume trigger management to any runtime.
@@ -168,7 +172,7 @@ async def _get_fired_triggers(self) -> dict[str, Any] | None:
                 UiPathResumeTriggerType.TIMER,
             )
         ]
-        return await self._build_resume_map(pollable_triggers)
+        return await self._build_resume_map(pollable_triggers, all_triggers=triggers)
 
     async def _restore_resume_input(
         self,
@@ -214,22 +218,32 @@ async def _restore_resume_input(
     async def _build_resume_map(
         self,
         triggers: list[UiPathResumeTrigger],
+        all_triggers: list[UiPathResumeTrigger] | None = None,
     ) -> dict[str, Any]:
         """Build resume map from triggers: {interrupt_id: resume_data}.
 
         Args:
             triggers: List of triggers to read and map
+            all_triggers: Full trigger set to use when detecting sibling triggers.
 
         Returns:
             A dict mapping interrupt_id to the trigger's resume data.
         """
         resume_map: dict[str, Any] = {}
+        trigger_count_by_interrupt_id = Counter(
+            trigger.interrupt_id for trigger in all_triggers or triggers
+        )
         for trigger in triggers:
+            assert trigger.interrupt_id is not None, (
+                "Trigger interrupt_id cannot be None"
+            )
+            if trigger.interrupt_id in resume_map:
+                continue
+
             try:
                 data = await self.trigger_manager.read_trigger(trigger)
-                assert trigger.interrupt_id is not None, (
-                    "Trigger interrupt_id cannot be None"
-                )
+                if trigger_count_by_interrupt_id[trigger.interrupt_id] > 1:
+                    data = self._with_trigger_metadata(trigger, data)
                 resume_map[trigger.interrupt_id] = data
                 await self.storage.delete_trigger(self.runtime_id, trigger)
             except UiPathPendingTriggerError:
@@ -238,6 +252,26 @@ async def _build_resume_map(
 
         return resume_map
 
+    def _with_trigger_metadata(
+        self, trigger: UiPathResumeTrigger, data: Any
+    ) -> dict[str, Any]:
+        metadata = {
+            "triggerType": self._metadata_value(trigger.trigger_type),
+            "triggerName": self._metadata_value(trigger.trigger_name),
+        }
+
+        if isinstance(data, dict):
+            return {**data, UIPATH_METADATA_KEY: metadata}
+
+        return {UIPATH_METADATA_KEY: metadata, "value": data}
+
+    @staticmethod
+    def _metadata_value(value: Any) -> Any:
+        if isinstance(value, Enum):
+            return value.value
+
+        return value
+
     async def _handle_suspension(
         self, result: UiPathRuntimeResult
     ) -> UiPathRuntimeResult:
@@ -273,11 +307,12 @@ async def _handle_suspension(
 
         # Create triggers only for new interrupts
         for interrupt_id in new_ids:
-            trigger = await self.trigger_manager.create_trigger(
+            triggers = await self.trigger_manager.create_triggers(
                 current_interrupts[interrupt_id]
             )
-            trigger.interrupt_id = interrupt_id
-            suspended_result.triggers.append(trigger)
+            for trigger in triggers:
+                trigger.interrupt_id = interrupt_id
+                suspended_result.triggers.append(trigger)
 
         if suspended_result.triggers:
             await self.storage.save_triggers(self.runtime_id, suspended_result.triggers)

tests/test_resumable.py

@@ -23,7 +23,7 @@
 from uipath.runtime.schema import UiPathRuntimeSchema
 
 
-class MultiTriggerMockRuntime:
+class SiblingTriggerMockRuntime:
     """Mock runtime that simulates parallel branching with multiple interrupts."""
 
     def __init__(self) -> None:
@@ -89,6 +89,56 @@ async def get_schema(self) -> UiPathRuntimeSchema:
         raise NotImplementedError()
 
 
+class MultiTriggerMockRuntime:
+    """Mock runtime that suspends on a multi-trigger interrupt, then suspends again."""
+
+    def __init__(self) -> None:
+        self.execution_count = 0
+
+    async def dispose(self) -> None:
+        pass
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        self.execution_count += 1
+        is_resume = options and options.resume
+
+        if self.execution_count == 1:
+            return UiPathRuntimeResult(
+                status=UiPathRuntimeStatus.SUSPENDED,
+                output={"child-1": {"process": "first-child"}},
+            )
+
+        assert is_resume
+        assert input == {
+            "child-1": {
+                "completed": True,
+                "__uipath": {
+                    "triggerType": UiPathResumeTriggerType.JOB.value,
+                    "triggerName": "Unknown",
+                },
+            }
+        }
+        return UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUSPENDED,
+            output={"child-2": {"process": "second-child"}},
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        result = await self.execute(input, options)
+        yield result
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        raise NotImplementedError()
+
+
 class StatefulStorageMock:
     """Stateful storage mock that tracks triggers."""
 
@@ -134,6 +184,11 @@ async def read_trigger_default(trigger: UiPathResumeTrigger) -> dict[str, Any]:
         raise UiPathPendingTriggerError(ErrorCategory.USER, "Trigger not fired yet")
 
     manager.create_trigger = AsyncMock(side_effect=create_trigger_impl)
+
+    async def create_triggers_impl(data: dict[str, Any]) -> list[UiPathResumeTrigger]:
+        return [await manager.create_trigger(data)]
+
+    manager.create_triggers = AsyncMock(side_effect=create_triggers_impl)
     manager.read_trigger = AsyncMock(side_effect=read_trigger_default)
 
     return cast(UiPathResumeTriggerProtocol, manager)
@@ -146,7 +201,7 @@ async def test_resumable_creates_multiple_triggers_on_first_suspension(
     ) -> None:
         """First suspension with parallel branches should create multiple triggers."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -179,7 +234,7 @@ async def test_resumable_creates_multiple_triggers_on_first_suspension(
     async def test_resumable_adds_only_new_triggers_on_partial_resume(self) -> None:
         """Partial resume should keep pending trigger and add only new ones."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -222,7 +277,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_completes_after_all_triggers_resolved(self) -> None:
         """After all triggers resolved, execution should complete successfully."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -269,7 +324,7 @@ async def read_trigger_impl_3(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_auto_resumes_when_triggers_already_fired(self) -> None:
         """When triggers are already fired during suspension, runtime should auto-resume."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -308,7 +363,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_auto_resumes_partial_fired_triggers(self) -> None:
         """When only some triggers are fired during suspension, auto-resume with those."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -344,7 +399,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_auto_resumes_multiple_times(self) -> None:
         """When triggers keep being fired immediately, keep auto-resuming until complete."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -389,7 +444,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_stream_auto_resumes_when_triggers_fired(self) -> None:
         """Stream should auto-resume when triggers are already fired."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -426,7 +481,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_no_auto_resume_when_all_triggers_pending(self) -> None:
         """When all triggers are pending, should NOT auto-resume."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -457,7 +512,7 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
     async def test_resumable_skips_api_triggers_on_auto_resume_check(self) -> None:
         """API triggers should be skipped when checking for auto-resume after suspension."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -508,7 +563,7 @@ async def test_resumable_skips_inbox_triggers_on_auto_resume_check(self) -> None
         404 and fault the run. They should behave like API triggers here.
         """
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -552,7 +607,7 @@ def create_inbox_trigger(data: dict[str, Any]) -> UiPathResumeTrigger:
     async def test_resumable_skips_timer_triggers_on_auto_resume_check(self) -> None:
         """Timer triggers should be skipped when checking for auto-resume."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -595,7 +650,7 @@ async def test_resumable_auto_resumes_task_triggers_but_not_api_triggers(
     ) -> None:
         """Mixed triggers: TASK triggers should trigger auto-resume, API triggers should not."""
 
-        runtime_impl = MultiTriggerMockRuntime()
+        runtime_impl = SiblingTriggerMockRuntime()
         storage = StatefulStorageMock()
         trigger_manager = make_trigger_manager_mock()
 
@@ -640,3 +695,67 @@ async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
 
         # Delegate should have been executed twice (initial + auto-resume for TASK trigger)
         assert runtime_impl.execution_count == 2
+
+    @pytest.mark.asyncio
+    async def test_resumable_removes_sibling_triggers_when_one_trigger_fires(
+        self,
+    ) -> None:
+        """When one trigger fires, sibling triggers must not leak forward."""
+
+        runtime_impl = MultiTriggerMockRuntime()
+        storage = StatefulStorageMock()
+        trigger_manager = make_trigger_manager_mock()
+
+        async def create_sibling_triggers(
+            data: dict[str, Any],
+        ) -> list[UiPathResumeTrigger]:
+            return [
+                UiPathResumeTrigger(
+                    interrupt_id="",  # Will be set by resumable runtime
+                    trigger_type=UiPathResumeTriggerType.JOB,
+                    payload={"source": data["process"]},
+                ),
+                UiPathResumeTrigger(
+                    interrupt_id="",  # Will be set by resumable runtime
+                    trigger_type=UiPathResumeTriggerType.TIMER,
+                    payload={"source": data["process"]},
+                ),
+            ]
+
+        create_triggers_mock = AsyncMock(side_effect=create_sibling_triggers)
+        cast(Any, trigger_manager).create_triggers = create_triggers_mock
+
+        async def read_trigger_impl(trigger: UiPathResumeTrigger) -> dict[str, Any]:
+            if (
+                trigger.interrupt_id == "child-1"
+                and trigger.trigger_type == UiPathResumeTriggerType.JOB
+            ):
+                return {"completed": True}
+            raise UiPathPendingTriggerError(ErrorCategory.USER, "still pending")
+
+        read_trigger_mock = AsyncMock(side_effect=read_trigger_impl)
+        cast(Any, trigger_manager).read_trigger = read_trigger_mock
+
+        resumable = UiPathResumableRuntime(
+            delegate=runtime_impl,
+            storage=storage,
+            trigger_manager=trigger_manager,
+            runtime_id="runtime-1",
+        )
+
+        result = await resumable.execute({})
+
+        assert result.status == UiPathRuntimeStatus.SUSPENDED
+        assert result.triggers is not None
+        assert {t.interrupt_id for t in result.triggers} == {"child-2"}
+        assert len(result.triggers) == 2
+        assert {t.trigger_type for t in result.triggers} == {
+            UiPathResumeTriggerType.JOB,
+            UiPathResumeTriggerType.TIMER,
+        }
+        assert runtime_impl.execution_count == 2
+        assert create_triggers_mock.await_count == 2
+        assert read_trigger_mock.await_count == 2
+
+        saved_triggers = await storage.get_triggers("runtime-1")
+        assert {t.interrupt_id for t in saved_triggers} == {"child-2"}