feat: add wait_for_triggers option and HeadlessDebugBridge for local polling

Add support for polling triggers instead of suspending, useful for running
agents locally without serverless/orchestrator.

## Option 1: Use wait_for_triggers flag (new)
Add `wait_for_triggers` and `trigger_poll_interval` options to UiPathExecuteOptions.
When enabled, triggers are created but NOT persisted to storage - everything
happens inline.

## Option 2: Use UiPathDebugRuntime with HeadlessDebugBridge (recommended by @cristipufu)
Add HeadlessDebugBridge - a no-op debug bridge for headless execution that allows
using UiPathDebugRuntime's existing polling feature without a debugger UI.

Changes:
- Add `wait_for_triggers` and `trigger_poll_interval` options to UiPathExecuteOptions
- Create shared TriggerPoller utility for reusable polling logic
- Update UiPathResumableRuntime to loop and poll when wait_for_triggers=True
- Add skip_storage parameter to _handle_suspension() to avoid persistence
- Add HeadlessDebugBridge for using UiPathDebugRuntime without a debugger
- Refactor UiPathDebugRuntime to use the shared TriggerPoller

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

Author: mjnovice

src/uipath/runtime/base.py

@@ -41,6 +41,16 @@ class UiPathExecuteOptions(BaseModel):
         default=None,
         description="List of nodes or '*' to break on all steps.",
     )
+    wait_for_triggers: bool = Field(
+        default=False,
+        description="When True, poll triggers until completion instead of suspending. "
+        "This keeps the process running and automatically resumes when triggers complete.",
+    )
+    trigger_poll_interval: float = Field(
+        default=5.0,
+        description="Seconds between poll attempts when wait_for_triggers is True.",
+        ge=0.1,
+    )
 
     model_config = {"arbitrary_types_allowed": True, "extra": "allow"}
 

src/uipath/runtime/debug/__init__.py

@@ -4,6 +4,7 @@
 from uipath.runtime.debug.exception import (
     UiPathDebugQuitError,
 )
+from uipath.runtime.debug.headless_bridge import HeadlessDebugBridge
 from uipath.runtime.debug.protocol import UiPathDebugProtocol
 from uipath.runtime.debug.runtime import UiPathDebugRuntime
 
@@ -12,4 +13,5 @@
     "UiPathDebugProtocol",
     "UiPathDebugRuntime",
     "UiPathBreakpointResult",
+    "HeadlessDebugBridge",
 ]

src/uipath/runtime/debug/headless_bridge.py

@@ -0,0 +1,120 @@
+"""Headless debug bridge for running without a debugger UI."""
+
+import asyncio
+from typing import Any, Literal
+
+from uipath.runtime.debug.breakpoint import UiPathBreakpointResult
+from uipath.runtime.events import UiPathRuntimeStateEvent
+from uipath.runtime.result import UiPathRuntimeResult
+
+
+class HeadlessDebugBridge:
+    """A no-op debug bridge for headless execution with polling.
+
+    Use this when you want to use UiPathDebugRuntime's polling feature
+    without an actual debugger UI connected.
+
+    Example:
+        ```python
+        from uipath.runtime.debug import UiPathDebugRuntime
+        from uipath.runtime.debug.headless_bridge import HeadlessDebugBridge
+
+        # Wrap your runtime with debug runtime using headless bridge
+        debug_runtime = UiPathDebugRuntime(
+            delegate=your_resumable_runtime,
+            debug_bridge=HeadlessDebugBridge(),
+            trigger_poll_interval=5.0,  # Poll every 5 seconds
+        )
+
+        # Execute - will poll triggers instead of suspending
+        result = await debug_runtime.execute(input={"query": "hello"})
+        ```
+    """
+
+    def __init__(self, verbose: bool = False):
+        """Initialize the headless debug bridge.
+
+        Args:
+            verbose: If True, print debug events to console
+        """
+        self.verbose = verbose
+        self._terminate_event = asyncio.Event()
+
+    async def connect(self) -> None:
+        """No-op connection."""
+        if self.verbose:
+            print("[HeadlessDebugBridge] Connected")
+
+    async def disconnect(self) -> None:
+        """No-op disconnection."""
+        if self.verbose:
+            print("[HeadlessDebugBridge] Disconnected")
+
+    async def emit_execution_started(self, **kwargs) -> None:
+        """Log execution started."""
+        if self.verbose:
+            print("[HeadlessDebugBridge] Execution started")
+
+    async def emit_state_update(self, state_event: UiPathRuntimeStateEvent) -> None:
+        """Log state update."""
+        if self.verbose:
+            print(f"[HeadlessDebugBridge] State update: {state_event.node_name}")
+
+    async def emit_breakpoint_hit(
+        self, breakpoint_result: UiPathBreakpointResult
+    ) -> None:
+        """Log breakpoint hit (should not happen in headless mode)."""
+        if self.verbose:
+            print(f"[HeadlessDebugBridge] Breakpoint hit: {breakpoint_result}")
+
+    async def emit_execution_suspended(
+        self, runtime_result: UiPathRuntimeResult
+    ) -> None:
+        """Log execution suspended."""
+        if self.verbose:
+            trigger_type = (
+                runtime_result.trigger.trigger_type
+                if runtime_result.trigger
+                else "unknown"
+            )
+            print(f"[HeadlessDebugBridge] Execution suspended, trigger: {trigger_type}")
+
+    async def emit_execution_resumed(self, resume_data: Any) -> None:
+        """Log execution resumed."""
+        if self.verbose:
+            print("[HeadlessDebugBridge] Execution resumed with data")
+
+    async def emit_execution_completed(
+        self,
+        runtime_result: UiPathRuntimeResult,
+    ) -> None:
+        """Log execution completed."""
+        if self.verbose:
+            print(f"[HeadlessDebugBridge] Execution completed: {runtime_result.status}")
+
+    async def emit_execution_error(
+        self,
+        error: str,
+    ) -> None:
+        """Log execution error."""
+        if self.verbose:
+            print(f"[HeadlessDebugBridge] Execution error: {error}")
+
+    async def wait_for_resume(self) -> Any:
+        """Return immediately to allow execution to continue."""
+        # In headless mode, we don't wait - just return immediately
+        return None
+
+    async def wait_for_terminate(self) -> None:
+        """Wait for termination signal (blocks forever in headless mode)."""
+        # This should never return in normal operation
+        # It's used to check if we should stop polling
+        await self._terminate_event.wait()
+
+    def terminate(self) -> None:
+        """Signal termination (call this to stop polling)."""
+        self._terminate_event.set()
+
+    def get_breakpoints(self) -> list[str] | Literal["*"]:
+        """Return no breakpoints for headless execution."""
+        return []

src/uipath/runtime/debug/runtime.py

@@ -4,8 +4,6 @@
 import logging
 from typing import Any, AsyncGenerator, cast
 
-from uipath.core.errors import UiPathPendingTriggerError
-
 from uipath.runtime.base import (
     UiPathExecuteOptions,
     UiPathRuntimeProtocol,
@@ -25,7 +23,7 @@
     UiPathRuntimeResult,
     UiPathRuntimeStatus,
 )
-from uipath.runtime.resumable.protocols import UiPathResumeTriggerReaderProtocol
+from uipath.runtime.resumable.polling import TriggerPoller
 from uipath.runtime.resumable.runtime import UiPathResumableRuntime
 from uipath.runtime.resumable.trigger import (
     UiPathResumeTrigger,
@@ -203,8 +201,7 @@ async def _stream_and_debug(
                                     )
                                 else:
                                     trigger_data = await self._poll_trigger(
-                                        final_result.trigger,
-                                        self.delegate.trigger_manager,
+                                        final_result.trigger
                                     )
                                 resume_data = {interrupt_id: trigger_data}
                             except UiPathDebugQuitError:
@@ -245,77 +242,65 @@ async def dispose(self) -> None:
             logger.warning(f"Error disconnecting debug bridge: {e}")
 
     async def _poll_trigger(
-        self, trigger: UiPathResumeTrigger, reader: UiPathResumeTriggerReaderProtocol
+        self, trigger: UiPathResumeTrigger
     ) -> dict[str, Any] | None:
         """Poll a resume trigger until data is available.
 
         Args:
             trigger: The trigger to poll
-            reader: The trigger reader to use for polling
 
         Returns:
-            Resume data when available, or None if polling exhausted
+            Resume data when available, or None if polling was stopped
 
         Raises:
             UiPathDebugQuitError: If quit is requested during polling
         """
-        attempt = 0
-        while True:
-            attempt += 1
-
-            try:
-                resume_data = await reader.read_trigger(trigger)
-
-                if resume_data is not None:
-                    return resume_data
-
-                await self.debug_bridge.emit_state_update(
-                    UiPathRuntimeStateEvent(
-                        node_name="<polling>",
-                        payload={
-                            "attempt": attempt,
-                        },
-                    )
+        self._quit_requested = False
+
+        async def on_poll_attempt(attempt: int, info: str | None) -> None:
+            """Callback for each poll attempt."""
+            payload: dict[str, Any] = {"attempt": attempt}
+            if info:
+                payload["info"] = info
+            await self.debug_bridge.emit_state_update(
+                UiPathRuntimeStateEvent(
+                    node_name="<polling>",
+                    payload=payload,
                 )
+            )
 
-                await self._wait_with_quit_check()
-
-            except UiPathDebugQuitError:
-                raise
-
-            except UiPathPendingTriggerError as e:
-                await self.debug_bridge.emit_state_update(
-                    UiPathRuntimeStateEvent(
-                        node_name="<polling>",
-                        payload={
-                            "attempt": attempt,
-                            "info": str(e),
-                        },
-                    )
+        async def should_stop() -> bool:
+            """Check if quit was requested."""
+            # Check for termination request with a short timeout
+            try:
+                term_task = asyncio.create_task(self.debug_bridge.wait_for_terminate())
+                done, _ = await asyncio.wait(
+                    {term_task},
+                    timeout=0.01,  # Very short timeout just to check
                 )
+                if term_task in done:
+                    self._quit_requested = True
+                    return True
+                else:
+                    term_task.cancel()
+                    try:
+                        await term_task
+                    except asyncio.CancelledError:
+                        pass
+            except Exception:
+                pass
+            return False
 
-                await self._wait_with_quit_check()
-
-    async def _wait_with_quit_check(self) -> None:
-        """Wait for specified seconds, but allow quit command to interrupt.
-
-        Raises:
-            UiPathDebugQuitError: If quit is requested during wait
-        """
-        sleep_task = asyncio.create_task(asyncio.sleep(self.trigger_poll_interval))
-        term_task = asyncio.create_task(self.debug_bridge.wait_for_terminate())
-
-        done, pending = await asyncio.wait(
-            {sleep_task, term_task},
-            return_when=asyncio.FIRST_COMPLETED,
+        poller = TriggerPoller(
+            reader=self.delegate.trigger_manager,
+            poll_interval=self.trigger_poll_interval,
+            on_poll_attempt=on_poll_attempt,
+            should_stop=should_stop,
         )
 
-        for task in pending:
-            task.cancel()
-            try:
-                await task
-            except asyncio.CancelledError:
-                pass
+        result = await poller.poll_trigger(trigger)
 
-        if term_task in done:
+        if self._quit_requested:
             raise UiPathDebugQuitError("Debugging terminated during polling.")
+
+        return result

src/uipath/runtime/resumable/__init__.py

@@ -1,5 +1,6 @@
 """Module for resumable runtime features."""
 
+from uipath.runtime.resumable.polling import TriggerPoller
 from uipath.runtime.resumable.protocols import (
     UiPathResumableStorageProtocol,
     UiPathResumeTriggerCreatorProtocol,
@@ -20,4 +21,5 @@
     "UiPathResumeTrigger",
     "UiPathResumeTriggerType",
     "UiPathApiTrigger",
+    "TriggerPoller",
 ]