fix: add mock template runtime with chat events

Author: cristipufu

src/uipath/dev/_demo/chat_agent/entry-points.json

@@ -1,12 +1,12 @@
 """Minimal demo script to run UiPathDevTerminal with mock runtimes."""
 
 import logging
+from pathlib import Path
 
 from uipath.runtime import (
     UiPathRuntimeProtocol,
 )
 
-from uipath.dev._demo.mock_context_runtime import ENTRYPOINT_CONTEXT, MockContextRuntime
 from uipath.dev._demo.mock_greeting_runtime import (
     ENTRYPOINT_GREETING,
     MockGreetingRuntime,
@@ -19,13 +19,32 @@
     ENTRYPOINT_SUPPORT_CHAT,
     MockSupportChatRuntime,
 )
+from uipath.dev._demo.mock_telemetry_runtime import (
+    ENTRYPOINT_TELEMETRY,
+    MockTelemetryRuntime,
+)
+from uipath.dev._demo.mock_template_runtime import (
+    create_template_runtime,
+)
 
 logger = logging.getLogger(__name__)
 
+# Template mappings: entrypoint -> (events_file, schema_file)
+TEMPLATE_RUNTIMES = {
+    "chat/movies.py:graph": (
+        "chat_agent/events.json",
+        "chat_agent/entry-points.json",
+    ),
+}
+
 
 class MockRuntimeFactory:
     """Runtime factory compatible with UiPathRuntimeFactoryProtocol."""
 
+    def __init__(self):
+        # Get the directory where this file is located (src/uipath/dev/_demo/)
+        self.demo_dir = Path(__file__).parent
+
     async def new_runtime(
         self, entrypoint: str, runtime_id: str
     ) -> UiPathRuntimeProtocol:
@@ -36,8 +55,19 @@ async def new_runtime(
             return MockNumberAnalyticsRuntime(entrypoint=entrypoint)
         if entrypoint == ENTRYPOINT_SUPPORT_CHAT:
             return MockSupportChatRuntime(entrypoint=entrypoint)
-        if entrypoint == ENTRYPOINT_CONTEXT:
-            return MockContextRuntime(entrypoint=entrypoint)
+        if entrypoint == ENTRYPOINT_TELEMETRY:
+            return MockTelemetryRuntime(entrypoint=entrypoint)
+
+        if entrypoint in TEMPLATE_RUNTIMES:
+            events_file, schema_file = TEMPLATE_RUNTIMES[entrypoint]
+
+            events_path = self.demo_dir / events_file
+            schema_path = self.demo_dir / schema_file
+
+            return create_template_runtime(
+                events_json=events_path,
+                schema_json=schema_path,
+            )
 
         # Fallback: still return something so the demo doesn't explode
         logger.warning(
@@ -47,20 +77,30 @@ async def new_runtime(
 
     async def discover_runtimes(self) -> list[UiPathRuntimeProtocol]:
         """Return prototype instances for discovery (not really used by the UI)."""
-        return [
+        runtimes = [
             MockGreetingRuntime(entrypoint=ENTRYPOINT_GREETING),
             MockNumberAnalyticsRuntime(entrypoint=ENTRYPOINT_ANALYZE_NUMBERS),
             MockSupportChatRuntime(entrypoint=ENTRYPOINT_SUPPORT_CHAT),
-            MockContextRuntime(entrypoint=ENTRYPOINT_CONTEXT),
+            MockTelemetryRuntime(entrypoint=ENTRYPOINT_TELEMETRY),
         ]
 
+        for entrypoint in TEMPLATE_RUNTIMES.keys():
+            try:
+                runtime = await self.new_runtime(entrypoint, runtime_id="discovery")
+                runtimes.append(runtime)
+            except Exception as e:
+                logger.error(f"Failed to load template runtime '{entrypoint}': {e}")
+
+        return runtimes
+
     def discover_entrypoints(self) -> list[str]:
         """Return all available entrypoints."""
         return [
-            ENTRYPOINT_CONTEXT,
+            ENTRYPOINT_TELEMETRY,
             ENTRYPOINT_GREETING,
             ENTRYPOINT_ANALYZE_NUMBERS,
             ENTRYPOINT_SUPPORT_CHAT,
+            *TEMPLATE_RUNTIMES.keys(),
         ]
 
     async def dispose(self) -> None:

src/uipath/dev/_demo/chat_agent/events.json

@@ -12,7 +12,7 @@
 )
 from uipath.runtime.schema import UiPathRuntimeSchema
 
-ENTRYPOINT_ANALYZE_NUMBERS = "analytics/numbers.py:analyze"
+ENTRYPOINT_ANALYZE_NUMBERS = "agent/numbers.py:analyze"
 
 logger = logging.getLogger(__name__)
 

src/uipath/dev/_demo/mock_runtime.py src/uipath/dev/_demo/mock_factory.pysrc/uipath/dev/_demo/mock_runtime.py renamed to src/uipath/dev/_demo/mock_factory.py

@@ -12,7 +12,7 @@
 )
 from uipath.runtime.schema import UiPathRuntimeSchema
 
-ENTRYPOINT_SUPPORT_CHAT = "agent/support.py:chat"
+ENTRYPOINT_SUPPORT_CHAT = "agent/support.py:main"
 
 logger = logging.getLogger(__name__)
 

src/uipath/dev/_demo/mock_numbers_runtime.py

@@ -12,15 +12,15 @@
 )
 from uipath.runtime.schema import UiPathRuntimeSchema
 
-ENTRYPOINT_CONTEXT = "agent/context.py:run"
+ENTRYPOINT_TELEMETRY = "agent/telemetry.py:run"
 
 logger = logging.getLogger(__name__)
 
 
-class MockContextRuntime:
+class MockTelemetryRuntime:
     """A mock runtime that simulates a multi-step workflow with rich telemetry."""
 
-    def __init__(self, entrypoint: str = ENTRYPOINT_CONTEXT) -> None:
+    def __init__(self, entrypoint: str = ENTRYPOINT_TELEMETRY) -> None:
         self.entrypoint = entrypoint
         self.tracer = trace.get_tracer("uipath.dev.mock.context")
         # State tracking for breakpoints

src/uipath/dev/_demo/mock_support_runtime.py

@@ -0,0 +1,200 @@
+"""Mock runtime that replays events from a JSON file."""
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+from typing import Any, AsyncGenerator
+
+from opentelemetry import trace
+from uipath.core.chat import UiPathConversationMessageEvent
+from uipath.runtime import (
+    UiPathExecuteOptions,
+    UiPathRuntimeEvent,
+    UiPathRuntimeResult,
+    UiPathRuntimeStatus,
+    UiPathStreamOptions,
+)
+from uipath.runtime.events import (
+    UiPathRuntimeMessageEvent,
+    UiPathRuntimeStateEvent,
+)
+from uipath.runtime.schema import (
+    UiPathRuntimeEdge,
+    UiPathRuntimeGraph,
+    UiPathRuntimeNode,
+    UiPathRuntimeSchema,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class MockTemplateRuntime:
+    """Mock runtime that replays events from a JSON file."""
+
+    def __init__(
+        self,
+        events_file: str | Path,
+        schema_file: str | Path,
+    ) -> None:
+        self.events_file = Path(events_file)
+        self.schema_file = Path(schema_file)
+        self.tracer = trace.get_tracer("uipath.dev.mock.template")
+        self._events: list[dict[str, Any]] = []
+        self._schema_data: dict[str, Any] | None = None
+        self._load_events()
+        self._load_schema()
+
+    def _load_events(self) -> None:
+        """Load events from JSON file."""
+        with open(self.events_file, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        if isinstance(data, list):
+            self._events = data
+        else:
+            raise ValueError("Expected JSON array of events")
+
+    def _load_schema(self) -> None:
+        """Load schema from separate JSON file if provided."""
+        with open(self.schema_file, "r", encoding="utf-8") as f:
+            self._schema_data = json.load(f)
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        """Get runtime schema from the loaded data or build default."""
+        if self._schema_data:
+            # Parse from entry-points.json format
+            entry_points = self._schema_data.get("entryPoints", [])
+            if entry_points:
+                ep = entry_points[0]
+
+                # Build graph if present
+                graph = None
+                if "graph" in ep and ep["graph"]:
+                    graph_data = ep["graph"]
+                    nodes = [
+                        UiPathRuntimeNode(**node)
+                        for node in graph_data.get("nodes", [])
+                    ]
+                    edges = [
+                        UiPathRuntimeEdge(**edge)
+                        for edge in graph_data.get("edges", [])
+                    ]
+                    graph = UiPathRuntimeGraph(nodes=nodes, edges=edges)
+
+                return UiPathRuntimeSchema(
+                    filePath=ep["filePath"],
+                    uniqueId=ep["uniqueId"],
+                    type=ep["type"],
+                    input=ep["input"],
+                    output=ep["output"],
+                    graph=graph,
+                )
+
+        # Fallback: return empty schema
+        return UiPathRuntimeSchema()
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        """Execute by replaying all events and returning final result."""
+        logger.info("MockTemplateRuntime: starting execution")
+
+        # Stream all events and capture the final result
+        final_result = None
+        async for event in self.stream(input=input, options=options):
+            if isinstance(event, UiPathRuntimeResult):
+                final_result = event
+
+        return final_result or UiPathRuntimeResult(
+            output={},
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        """Stream events from the JSON file."""
+        logger.info(f"MockTemplateRuntime: streaming {len(self._events)} events")
+
+        with self.tracer.start_as_current_span(
+            "template.stream",
+            attributes={
+                "uipath.runtime.name": "MockTemplateRuntime",
+                "uipath.event.count": len(self._events),
+            },
+        ):
+            for i, event_data in enumerate(self._events):
+                event_type = event_data.get("event_type")
+
+                # Add small delay between events for realistic streaming
+                if i > 0:
+                    await asyncio.sleep(0.01)
+
+                try:
+                    if event_type == "runtime_message":
+                        payload_data = event_data.get("payload", {})
+
+                        conversation_event = (
+                            UiPathConversationMessageEvent.model_validate(payload_data)
+                        )
+
+                        yield UiPathRuntimeMessageEvent(
+                            payload=conversation_event,
+                            execution_id=event_data.get("execution_id"),
+                            metadata=event_data.get("metadata"),
+                        )
+
+                    elif event_type == "runtime_state":
+                        yield UiPathRuntimeStateEvent(
+                            payload=event_data.get("payload", {}),
+                            node_name=event_data.get("node_name"),
+                            execution_id=event_data.get("execution_id"),
+                            metadata=event_data.get("metadata"),
+                        )
+
+                    elif event_type == "runtime_result":
+                        yield UiPathRuntimeResult(
+                            output=event_data.get("output"),
+                            status=UiPathRuntimeStatus(
+                                event_data.get("status", "successful")
+                            ),
+                            execution_id=event_data.get("execution_id"),
+                            metadata=event_data.get("metadata"),
+                        )
+
+                    else:
+                        logger.warning(f"Unknown event type: {event_type}")
+
+                except Exception as e:
+                    logger.error(f"Error processing event {i}: {e}", exc_info=True)
+                    continue
+
+        logger.info("MockReplayRuntime: streaming completed")
+
+    async def dispose(self) -> None:
+        """Cleanup resources."""
+        logger.info("MockReplayRuntime: dispose() invoked")
+
+
+def create_template_runtime(
+    events_json: str | Path,
+    schema_json: str | Path | None = None,
+) -> MockTemplateRuntime:
+    """Create a template runtime from JSON files.
+
+    Args:
+        events_json: Path to JSON file containing events array
+        schema_json: Optional path to entry-points.json schema file
+
+    Returns:
+        MockTemplateRuntime instance
+    """
+    return MockTemplateRuntime(
+        events_file=events_json,
+        schema_file=schema_json,
+    )

…uipath/dev/_demo/mock_context_runtime.py …path/dev/_demo/mock_telemetry_runtime.pysrc/uipath/dev/_demo/mock_context_runtime.py renamed to src/uipath/dev/_demo/mock_telemetry_runtime.py src/uipath/dev/_demo/mock_context_runtime.py renamed to src/uipath/dev/_demo/mock_telemetry_runtime.py

@@ -1,7 +1,7 @@
 from uipath.core.tracing import UiPathTraceManager
 
 from uipath.dev import UiPathDeveloperConsole
-from uipath.dev._demo.mock_runtime import MockRuntimeFactory
+from uipath.dev._demo.mock_factory import MockRuntimeFactory
 
 
 def main():