fix: treat timer triggers as external resume triggers

pyproject.toml

@@ -1,11 +1,11 @@
 [project]
 name = "uipath-runtime"
-version = "0.11.4"
+version = "0.11.5"
 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"
 dependencies = [
-  "uipath-core>=0.5.22,<0.6.0",
+  "uipath-core>=0.5.26,<0.6.0",
 ]
 classifiers = [
   "Intended Audience :: Developers",

src/uipath/runtime/debug/runtime.py

@@ -194,9 +194,10 @@ async def _stream_and_debug(
                             resume_data: dict[str, Any] | None = None
                             try:
                                 trigger_data: dict[str, Any] | None = None
-                                if (
-                                    final_result.trigger.trigger_type
-                                    == UiPathResumeTriggerType.API
+                                if final_result.trigger.trigger_type in (
+                                    UiPathResumeTriggerType.API,
+                                    UiPathResumeTriggerType.INBOX,
+                                    UiPathResumeTriggerType.TIMER,
                                 ):
                                     trigger_data = (
                                         await self.debug_bridge.wait_for_resume()

src/uipath/runtime/resumable/runtime.py

@@ -147,8 +147,9 @@ async def stream(
     async def _get_fired_triggers(self) -> dict[str, Any] | None:
         """Check stored triggers for any that have already fired.
 
-        Skips async-external triggers (API, Inbox) whose payloads only arrive
-        asynchronously and cannot be polled at suspend time.
+        Skips external triggers (API, Inbox, Timer) whose payloads only arrive
+        asynchronously or through Orchestrator resume and cannot be polled at
+        suspend time.
 
         Returns:
             A resume map of {interrupt_id: resume_data} for fired triggers, or None.
@@ -161,7 +162,11 @@ async def _get_fired_triggers(self) -> dict[str, Any] | None:
             t
             for t in triggers
             if t.trigger_type
-            not in (UiPathResumeTriggerType.API, UiPathResumeTriggerType.INBOX)
+            not in (
+                UiPathResumeTriggerType.API,
+                UiPathResumeTriggerType.INBOX,
+                UiPathResumeTriggerType.TIMER,
+            )
         ]
         return await self._build_resume_map(pollable_triggers)
 

tests/test_debugger.py

@@ -6,6 +6,7 @@
 from unittest.mock import AsyncMock, Mock
 
 import pytest
+from uipath.core.triggers import UiPathResumeTrigger, UiPathResumeTriggerType
 
 from uipath.runtime import (
     UiPathBreakpointResult,
@@ -131,6 +132,48 @@ async def get_schema(self) -> UiPathRuntimeSchema:
         raise NotImplementedError()
 
 
+class SuspendedThenSuccessfulRuntime:
+    """Mock runtime that suspends once and completes after resume."""
+
+    def __init__(self, trigger: UiPathResumeTrigger) -> None:
+        self.trigger = trigger
+        self.inputs: list[dict[str, Any] | None] = []
+        self.options: list[UiPathStreamOptions | None] = []
+
+    async def dispose(self) -> None:
+        pass
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        raise NotImplementedError()
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        self.inputs.append(input)
+        self.options.append(options)
+
+        if options and options.resume:
+            yield UiPathRuntimeResult(
+                status=UiPathRuntimeStatus.SUCCESSFUL,
+                output={"resumed_with": input},
+            )
+            return
+
+        yield UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUSPENDED,
+            trigger=self.trigger,
+        )
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        raise NotImplementedError()
+
+
 @pytest.mark.asyncio
 async def test_debug_runtime_streams_and_handles_breakpoints_and_state():
     """UiPathDebugRuntime should stream events, handle breakpoints and state updates."""
@@ -169,6 +212,51 @@ async def test_debug_runtime_streams_and_handles_breakpoints_and_state():
     )  # initial + after breakpoint
 
 
+@pytest.mark.asyncio
+async def test_debug_runtime_waits_for_timer_resume_without_polling():
+    """Timer triggers should wait for external resume in debug mode."""
+    trigger = UiPathResumeTrigger(
+        interrupt_id="timer-interrupt",
+        trigger_type=UiPathResumeTriggerType.TIMER,
+        payload={"kind": "timeout"},
+    )
+    runtime_impl = SuspendedThenSuccessfulRuntime(trigger)
+    bridge = make_debug_bridge_mock()
+    cast(AsyncMock, bridge.wait_for_resume).side_effect = [
+        None,
+        {"__uipath": {"kind": "timeout"}},
+    ]
+    cast(Mock, bridge.get_breakpoints).return_value = []
+
+    trigger_manager = Mock()
+    trigger_manager.read_trigger = AsyncMock(
+        side_effect=AssertionError("Timer triggers must not be polled")
+    )
+
+    debug_runtime = UiPathDebugRuntime(
+        delegate=runtime_impl,
+        debug_bridge=bridge,
+    )
+    debug_runtime.get_resumable_runtime = Mock(  # type: ignore[method-assign]
+        return_value=Mock(trigger_manager=trigger_manager)
+    )
+
+    result = await debug_runtime.execute({})
+
+    assert result.status == UiPathRuntimeStatus.SUCCESSFUL
+    assert result.output == {
+        "resumed_with": {
+            "timer-interrupt": {"__uipath": {"kind": "timeout"}},
+        },
+    }
+    assert cast(AsyncMock, bridge.wait_for_resume).await_count == 2
+    cast(AsyncMock, bridge.emit_execution_suspended).assert_awaited_once()
+    cast(AsyncMock, bridge.emit_execution_resumed).assert_awaited_once_with(
+        {"timer-interrupt": {"__uipath": {"kind": "timeout"}}}
+    )
+    trigger_manager.read_trigger.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_debug_runtime_falls_back_when_stream_not_supported():
     """If runtime raises UiPathStreamNotSupportedError, we fall back to execute()."""

tests/test_resumable.py

@@ -548,6 +548,47 @@ def create_inbox_trigger(data: dict[str, Any]) -> UiPathResumeTrigger:
         trigger_manager.read_trigger.assert_not_called()
         assert runtime_impl.execution_count == 1
 
+    @pytest.mark.asyncio
+    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()
+        storage = StatefulStorageMock()
+        trigger_manager = make_trigger_manager_mock()
+
+        def create_timer_trigger(data: dict[str, Any]) -> UiPathResumeTrigger:
+            return UiPathResumeTrigger(
+                interrupt_id="",  # Will be set by resumable runtime
+                trigger_type=UiPathResumeTriggerType.TIMER,
+                payload=data,
+            )
+
+        trigger_manager.create_trigger = AsyncMock(side_effect=create_timer_trigger)  # type: ignore[method-assign]
+        read_trigger_guard = AsyncMock(
+            side_effect=AssertionError(
+                "read_trigger must not be called for Timer triggers pre-resume"
+            )
+        )
+        trigger_manager.read_trigger = read_trigger_guard  # type: ignore[method-assign]
+
+        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 len(result.triggers) == 2
+        assert all(
+            t.trigger_type == UiPathResumeTriggerType.TIMER for t in result.triggers
+        )
+        trigger_manager.read_trigger.assert_not_called()
+        assert runtime_impl.execution_count == 1
+
     @pytest.mark.asyncio
     async def test_resumable_auto_resumes_task_triggers_but_not_api_triggers(
         self,