feat: propagate errors from agent runtime to CAS

Author: andrewwan-uipath

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.9.3"
+version = "0.9.4"
 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/chat/protocol.py

@@ -47,6 +47,12 @@ async def emit_exchange_end_event(self) -> None:
         """Send an exchange end event."""
         ...
 
+    async def emit_exchange_error_event(
+        self, error_id: str, message: str, details: Any | None = None
+    ) -> None:
+        """Emit an exchange error event."""
+        ...
+
     async def wait_for_resume(self) -> dict[str, Any]:
         """Wait for the interrupt_end event to be received."""
         ...

src/uipath/runtime/chat/runtime.py

@@ -11,6 +11,7 @@
     UiPathStreamOptions,
 )
 from uipath.runtime.chat.protocol import UiPathChatProtocol
+from uipath.runtime.errors import UiPathBaseRuntimeError, UiPathErrorContract
 from uipath.runtime.events import (
     UiPathRuntimeEvent,
     UiPathRuntimeMessageEvent,
@@ -23,6 +24,32 @@
 
 logger = logging.getLogger(__name__)
 
+_DEFAULT_ERROR_ID = "AGENT_RUNTIME_ERROR"
+_DEFAULT_ERROR_MESSAGE = "An unexpected error occurred."
+
+
+def _extract_error_from_exception(e: Exception) -> tuple[str, str]:
+    """Extract error_id and user-facing message from an exception."""
+    if isinstance(e, UiPathBaseRuntimeError):
+        return _extract_error_from_contract(e.error_info)
+    return _DEFAULT_ERROR_ID, _DEFAULT_ERROR_MESSAGE
+
+
+def _extract_error_from_contract(
+    error: UiPathErrorContract | None,
+) -> tuple[str, str]:
+    """Extract error_id and user-facing message from an error contract."""
+    if not error:
+        return _DEFAULT_ERROR_ID, _DEFAULT_ERROR_MESSAGE
+    error_id = error.code or _DEFAULT_ERROR_ID
+    title = error.title or ""
+    detail = error.detail.split("\n")[0] if error.detail else ""
+    if title and detail:
+        error_message = f"{title}. {detail}"
+    else:
+        error_message = title or detail or _DEFAULT_ERROR_MESSAGE
+    return error_id, error_message
+
 
 class UiPathChatRuntime:
     """Specialized runtime for chat mode that streams message events to a chat bridge."""
@@ -65,62 +92,86 @@ async def stream(
         options: UiPathStreamOptions | None = None,
     ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
         """Stream execution events with chat support."""
-        await self.chat_bridge.connect()
-
-        execution_completed = False
-        current_input = input
-        current_options = UiPathStreamOptions(
-            resume=options.resume if options else False,
-            breakpoints=options.breakpoints if options else None,
-        )
-
-        while not execution_completed:
-            async for event in self.delegate.stream(
-                current_input, options=current_options
-            ):
-                if isinstance(event, UiPathRuntimeMessageEvent):
-                    if event.payload:
-                        await self.chat_bridge.emit_message_event(event.payload)
-
-                if isinstance(event, UiPathRuntimeResult):
-                    runtime_result = event
-
-                    if (
-                        runtime_result.status == UiPathRuntimeStatus.SUSPENDED
-                        and runtime_result.triggers
-                    ):
-                        api_triggers = [
-                            t
-                            for t in runtime_result.triggers
-                            if t.trigger_type == UiPathResumeTriggerType.API
-                        ]
-
-                        if api_triggers:
-                            resume_map: dict[str, Any] = {}
-
-                            for trigger in api_triggers:
-                                await self.chat_bridge.emit_interrupt_event(trigger)
-
-                                resume_data = await self.chat_bridge.wait_for_resume()
-
-                                assert trigger.interrupt_id is not None, (
-                                    "Trigger interrupt_id cannot be None"
-                                )
-                                resume_map[trigger.interrupt_id] = resume_data
-
-                            current_input = resume_map
-                            current_options.resume = True
-                            break
+        try:
+            await self.chat_bridge.connect()
+
+            execution_completed = False
+            current_input = input
+            current_options = UiPathStreamOptions(
+                resume=options.resume if options else False,
+                breakpoints=options.breakpoints if options else None,
+            )
+
+            while not execution_completed:
+                async for event in self.delegate.stream(
+                    current_input, options=current_options
+                ):
+                    if isinstance(event, UiPathRuntimeMessageEvent):
+                        if event.payload:
+                            await self.chat_bridge.emit_message_event(event.payload)
+
+                    if isinstance(event, UiPathRuntimeResult):
+                        runtime_result = event
+
+                        if (
+                            runtime_result.status == UiPathRuntimeStatus.SUSPENDED
+                            and runtime_result.triggers
+                        ):
+                            api_triggers = [
+                                t
+                                for t in runtime_result.triggers
+                                if t.trigger_type == UiPathResumeTriggerType.API
+                            ]
+
+                            if api_triggers:
+                                resume_map: dict[str, Any] = {}
+
+                                for trigger in api_triggers:
+                                    await self.chat_bridge.emit_interrupt_event(trigger)
+
+                                    resume_data = (
+                                        await self.chat_bridge.wait_for_resume()
+                                    )
+
+                                    assert trigger.interrupt_id is not None, (
+                                        "Trigger interrupt_id cannot be None"
+                                    )
+                                    resume_map[trigger.interrupt_id] = resume_data
+
+                                current_input = resume_map
+                                current_options.resume = True
+                                break
+                            else:
+                                # No API triggers - yield result and complete
+                                yield event
+                                execution_completed = True
+                        elif runtime_result.status == UiPathRuntimeStatus.FAULTED:
+                            await self._emit_error_event(
+                                *_extract_error_from_contract(runtime_result.error)
+                            )
+                            yield event
+                            execution_completed = True
                         else:
-                            # No API triggers - yield result and complete
                             yield event
                             execution_completed = True
+                            await self.chat_bridge.emit_exchange_end_event()
                     else:
                         yield event
-                        execution_completed = True
-                        await self.chat_bridge.emit_exchange_end_event()
-                else:
-                    yield event
+
+        except Exception as e:
+            error_id, error_message = _extract_error_from_exception(e)
+            await self._emit_error_event(error_id, error_message)
+            raise
+
+    async def _emit_error_event(self, error_id: str, message: str) -> None:
+        """Emit an exchange error event to the chat bridge."""
+        try:
+            await self.chat_bridge.emit_exchange_error_event(
+                error_id=error_id,
+                message=message,
+            )
+        except Exception:
+            logger.warning("Failed to emit exchange error event", exc_info=True)
 
     async def get_schema(self) -> UiPathRuntimeSchema:
         """Get schema from the delegate runtime."""