feat: add detached debug bridge for non-interactive runs

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.10.0"
+version = "0.10.1"
 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/debug/__init__.py

@@ -1,13 +1,15 @@
 """Initialization module for the debug package."""
 
 from uipath.runtime.debug.breakpoint import UiPathBreakpointResult
+from uipath.runtime.debug.detached import DetachedDebugBridge
 from uipath.runtime.debug.exception import (
     UiPathDebugQuitError,
 )
 from uipath.runtime.debug.protocol import UiPathDebugProtocol
 from uipath.runtime.debug.runtime import UiPathDebugRuntime
 
 __all__ = [
+    "DetachedDebugBridge",
     "UiPathDebugQuitError",
     "UiPathDebugProtocol",
     "UiPathDebugRuntime",

src/uipath/runtime/debug/detached.py

@@ -0,0 +1,73 @@
+"""Detached debug bridge — satisfies `UiPathDebugProtocol` without attaching a debugger."""
+
+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 DetachedDebugBridge:
+    """Debug bridge used when no debugger is attached.
+
+    Implements `UiPathDebugProtocol` so the debug runtime stack keeps wrapping
+    uniformly, but all hooks are no-ops. `wait_for_resume` returns immediately
+    so the runtime's initial paused-state gate releases without blocking;
+    `wait_for_terminate` blocks forever because termination can never arrive
+    through a bridge that isn't connected to anything.
+    """
+
+    async def connect(self) -> None:
+        """No-op — nothing to connect to when detached."""
+        pass
+
+    async def disconnect(self) -> None:
+        """No-op — no connection to tear down."""
+        pass
+
+    async def emit_execution_started(self, **kwargs: Any) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_state_update(self, state_event: UiPathRuntimeStateEvent) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_breakpoint_hit(
+        self, breakpoint_result: UiPathBreakpointResult
+    ) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_execution_suspended(
+        self, runtime_result: UiPathRuntimeResult
+    ) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_execution_resumed(self, resume_data: Any) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_execution_completed(
+        self, runtime_result: UiPathRuntimeResult
+    ) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def emit_execution_error(self, error: str) -> None:
+        """No-op — no debugger is listening."""
+        pass
+
+    async def wait_for_resume(self) -> Any:
+        """Return immediately — the runtime's initial paused gate releases without a debugger."""
+        return None
+
+    async def wait_for_terminate(self) -> None:
+        """Block forever — termination cannot arrive when no debugger is attached."""
+        await asyncio.Event().wait()
+
+    def get_breakpoints(self) -> list[str] | Literal["*"]:
+        """Return an empty breakpoint list so the runtime never suspends."""
+        return []

tests/test_detached_debug_bridge.py

@@ -0,0 +1,77 @@
+"""Tests for DetachedDebugBridge."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from uipath.runtime.debug import (
+    DetachedDebugBridge,
+    UiPathBreakpointResult,
+    UiPathDebugProtocol,
+)
+from uipath.runtime.events import UiPathRuntimeStateEvent
+from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus
+
+
+def test_detached_bridge_satisfies_debug_protocol():
+    """DetachedDebugBridge must be usable wherever UiPathDebugProtocol is expected."""
+    bridge: UiPathDebugProtocol = DetachedDebugBridge()
+    assert bridge is not None
+
+
+@pytest.mark.asyncio
+async def test_connect_and_disconnect_are_noops():
+    """Lifecycle methods must complete without raising."""
+    bridge = DetachedDebugBridge()
+    await bridge.connect()
+    await bridge.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_all_emit_methods_are_noops():
+    """Emit methods must not raise and must not require any external state."""
+    bridge = DetachedDebugBridge()
+
+    state_event = UiPathRuntimeStateEvent(node_name="node-x", payload={})
+    breakpoint_result = UiPathBreakpointResult(
+        breakpoint_node="node-x",
+        breakpoint_type="before",
+        next_nodes=[],
+        current_state={},
+    )
+    runtime_result = UiPathRuntimeResult(
+        status=UiPathRuntimeStatus.SUCCESSFUL,
+        output={},
+    )
+
+    await bridge.emit_execution_started()
+    await bridge.emit_state_update(state_event)
+    await bridge.emit_breakpoint_hit(breakpoint_result)
+    await bridge.emit_execution_suspended(runtime_result)
+    await bridge.emit_execution_resumed({"any": "data"})
+    await bridge.emit_execution_completed(runtime_result)
+    await bridge.emit_execution_error("boom")
+
+
+@pytest.mark.asyncio
+async def test_wait_for_resume_returns_immediately():
+    """The runtime's initial paused gate calls this — it must release without blocking."""
+    bridge = DetachedDebugBridge()
+    # Fails the test if the call hangs for any reason.
+    await asyncio.wait_for(bridge.wait_for_resume(), timeout=1.0)
+
+
+@pytest.mark.asyncio
+async def test_wait_for_terminate_blocks_forever():
+    """Termination can never arrive on a detached bridge — the coroutine must not complete."""
+    bridge = DetachedDebugBridge()
+    with pytest.raises(asyncio.TimeoutError):
+        await asyncio.wait_for(bridge.wait_for_terminate(), timeout=0.1)
+
+
+def test_get_breakpoints_returns_empty_list():
+    """Empty list means 'no breakpoints' — the runtime's normal flow then skips suspension."""
+    bridge = DetachedDebugBridge()
+    assert bridge.get_breakpoints() == []

tests/test_detached_debug_bridge_integration.py

@@ -0,0 +1,122 @@
+"""Integration test: UiPathDebugRuntime must not block under DetachedDebugBridge.
+
+If this ever hangs or times out, the detached path has regressed — the scenario
+this bridge exists to enable has broken.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncGenerator
+
+import pytest
+
+from uipath.runtime import (
+    UiPathExecuteOptions,
+    UiPathRuntimeResult,
+    UiPathRuntimeStatus,
+    UiPathStreamNotSupportedError,
+    UiPathStreamOptions,
+)
+from uipath.runtime.debug import DetachedDebugBridge, UiPathDebugRuntime
+from uipath.runtime.events import UiPathRuntimeEvent, UiPathRuntimeStateEvent
+from uipath.runtime.schema import UiPathRuntimeSchema
+
+
+class TrivialStreamingRuntime:
+    """Streams one state event then a final successful result."""
+
+    async def dispose(self) -> None:
+        pass
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        return UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"mode": "execute"},
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        yield UiPathRuntimeStateEvent(node_name="node-1", payload={"i": 0})
+        yield UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"done": True},
+        )
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        raise NotImplementedError()
+
+
+class NonStreamingRuntime:
+    """Raises UiPathStreamNotSupportedError — forces the execute() fallback path."""
+
+    def __init__(self) -> None:
+        self.execute_called = False
+
+    async def dispose(self) -> None:
+        pass
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        self.execute_called = True
+        return UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"mode": "execute"},
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        raise UiPathStreamNotSupportedError("nope")
+        yield  # pragma: no cover — makes this an async generator
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        raise NotImplementedError()
+
+
+@pytest.mark.asyncio
+async def test_debug_runtime_streams_to_completion_under_detached_bridge():
+    """The detached bridge must not block the runtime's startup wait-for-resume gate."""
+    debug_runtime = UiPathDebugRuntime(
+        delegate=TrivialStreamingRuntime(),
+        debug_bridge=DetachedDebugBridge(),
+    )
+
+    try:
+        result = await asyncio.wait_for(debug_runtime.execute({}), timeout=5.0)
+    finally:
+        await debug_runtime.dispose()
+
+    assert isinstance(result, UiPathRuntimeResult)
+    assert result.status == UiPathRuntimeStatus.SUCCESSFUL
+    assert result.output == {"done": True}
+
+
+@pytest.mark.asyncio
+async def test_debug_runtime_execute_fallback_completes_under_detached_bridge():
+    """Fallback path (stream-unsupported delegates) must also not block."""
+    delegate = NonStreamingRuntime()
+    debug_runtime = UiPathDebugRuntime(
+        delegate=delegate,
+        debug_bridge=DetachedDebugBridge(),
+    )
+
+    try:
+        result = await asyncio.wait_for(debug_runtime.execute({}), timeout=5.0)
+    finally:
+        await debug_runtime.dispose()
+
+    assert delegate.execute_called is True
+    assert result.status == UiPathRuntimeStatus.SUCCESSFUL