fix: make LiveTrackingSpanProcessor upsert calls non-blocking (#1116)

Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: AAgnihotry

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.4.22"
+version = "2.4.23"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath/_cli/_evals/_runtime.py

@@ -2,6 +2,7 @@
 import logging
 import uuid
 from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from pathlib import Path
 from time import time
@@ -166,33 +167,59 @@ class LiveTrackingSpanProcessor(SpanProcessor):
     Sends real-time span updates:
     - On span start: Upsert with RUNNING status
     - On span end: Upsert with final status (OK/ERROR)
+
+    All upsert calls run in background threads without blocking evaluation
+    execution. Uses a thread pool to cap the maximum number of concurrent
+    threads and avoid resource exhaustion.
     """
 
-    def __init__(self, exporter: LlmOpsHttpExporter):
+    def __init__(
+        self,
+        exporter: LlmOpsHttpExporter,
+        max_workers: int = 10,
+    ):
         self.exporter = exporter
         self.span_status = SpanStatus
+        self.executor = ThreadPoolExecutor(
+            max_workers=max_workers, thread_name_prefix="span-upsert"
+        )
+
+    def _upsert_span_async(
+        self, span: Span | ReadableSpan, status_override: int | None = None
+    ) -> None:
+        """Run upsert_span in a background thread without blocking.
+
+        Submits the upsert task to the thread pool and returns immediately.
+        The thread pool handles execution with max_workers cap to prevent
+        resource exhaustion.
+        """
+
+        def _upsert():
+            try:
+                if status_override:
+                    self.exporter.upsert_span(span, status_override=status_override)
+                else:
+                    self.exporter.upsert_span(span)
+            except Exception as e:
+                logger.debug(f"Failed to upsert span: {e}")
+
+        # Submit to thread pool and return immediately (non-blocking)
+        # The timeout parameter is reserved for shutdown operations
+        self.executor.submit(_upsert)
 
     def on_start(
         self, span: Span, parent_context: context_api.Context | None = None
     ) -> None:
-        """Called when span starts - upsert with RUNNING status."""
+        """Called when span starts - upsert with RUNNING status (non-blocking)."""
         # Only track evaluation-related spans
         if span.attributes and self._is_eval_span(span):
-            try:
-                self.exporter.upsert_span(
-                    span, status_override=self.span_status.RUNNING
-                )
-            except Exception as e:
-                logger.debug(f"Failed to upsert span on start: {e}")
+            self._upsert_span_async(span, status_override=self.span_status.RUNNING)
 
     def on_end(self, span: ReadableSpan) -> None:
-        """Called when span ends - upsert with final status."""
+        """Called when span ends - upsert with final status (non-blocking)."""
         # Only track evaluation-related spans
         if span.attributes and self._is_eval_span(span):
-            try:
-                self.exporter.upsert_span(span)
-            except Exception as e:
-                logger.debug(f"Failed to upsert span on end: {e}")
+            self._upsert_span_async(span)
 
     def _is_eval_span(self, span: Span | ReadableSpan) -> bool:
         """Check if span is evaluation-related."""
@@ -219,8 +246,11 @@ def _is_eval_span(self, span: Span | ReadableSpan) -> bool:
         return False
 
     def shutdown(self) -> None:
-        """Shutdown the processor."""
-        pass
+        """Shutdown the processor and wait for pending tasks to complete."""
+        try:
+            self.executor.shutdown(wait=True)
+        except Exception as e:
+            logger.debug(f"Executor shutdown failed: {e}")
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         """Force flush - no-op for live tracking."""

tests/cli/eval/test_live_tracking_span_processor.py

@@ -1,5 +1,7 @@
 """Tests for LiveTrackingSpanProcessor in _runtime.py."""
 
+import threading
+import time
 from typing import Any
 from unittest.mock import Mock
 
@@ -290,3 +292,208 @@ def test_processor_handles_all_eval_span_types(self, processor):
             assert processor._is_eval_span(span) is True, (
                 f"Failed for span_type: {span_type}"
             )
+
+    # Tests for ThreadPoolExecutor behavior
+
+    def test_thread_pool_executor_used(self, mock_exporter):
+        """Test that processor uses ThreadPoolExecutor for async operations."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        # Verify executor exists
+        assert hasattr(processor, "executor")
+        assert processor.executor is not None
+
+        # Submit task and verify it's non-blocking
+        start_time = time.time()
+        processor.on_start(span, None)
+        elapsed = time.time() - start_time
+
+        # Should return immediately (< 0.05 seconds)
+        assert elapsed < 0.05, f"on_start blocked for {elapsed} seconds"
+
+    def test_handles_exceptions_gracefully(self, mock_exporter):
+        """Test that exceptions in background threads don't crash."""
+        mock_exporter.upsert_span = Mock(side_effect=Exception("Network error"))
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        # Should not raise exception
+        processor.on_start(span, None)
+        # Wait for background thread to process
+        time.sleep(0.2)
+
+        # Main thread should still be alive
+        assert threading.current_thread().is_alive()
+
+    def test_upsert_span_async_with_status_override(self, mock_exporter):
+        """Test _upsert_span_async correctly passes status_override."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        processor._upsert_span_async(span, status_override=SpanStatus.RUNNING)
+
+        # Wait for background thread to complete
+        time.sleep(0.2)
+
+        mock_exporter.upsert_span.assert_called_once_with(
+            span, status_override=SpanStatus.RUNNING
+        )
+
+    def test_upsert_span_async_without_status_override(self, mock_exporter):
+        """Test _upsert_span_async without status_override."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_readable_span({"span_type": "eval"})
+
+        processor._upsert_span_async(span, status_override=None)
+
+        # Wait for background thread to complete
+        time.sleep(0.2)
+
+        mock_exporter.upsert_span.assert_called_once_with(span)
+
+    def test_processor_initialization_with_custom_max_workers(self, mock_exporter):
+        """Test processor can be initialized with custom max_workers."""
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=15)
+        assert processor.executor._max_workers == 15
+
+    def test_exception_in_background_thread_does_not_crash(self, mock_exporter):
+        """Test that exceptions in background threads don't crash the main thread."""
+        mock_exporter.upsert_span = Mock(side_effect=Exception("Background error"))
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        # Should not raise exception
+        processor.on_start(span, None)
+        time.sleep(0.2)  # Wait for background thread
+
+        # Main thread should still be alive
+        assert threading.current_thread().is_alive()
+
+    # Tests for ThreadPoolExecutor and max_workers
+
+    def test_processor_with_custom_max_workers(self, mock_exporter):
+        """Test processor can be initialized with custom max_workers."""
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=20)
+        assert processor.executor._max_workers == 20
+
+    def test_processor_default_max_workers(self, mock_exporter):
+        """Test processor uses default max_workers of 10."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        assert processor.executor._max_workers == 10
+
+    def test_thread_pool_caps_concurrent_threads(self, mock_exporter):
+        """Test that thread pool caps concurrent threads to max_workers."""
+        concurrent_calls = []
+        max_concurrent = 0
+
+        def slow_upsert(*args, **kwargs):
+            concurrent_calls.append(1)
+            nonlocal max_concurrent
+            max_concurrent = max(max_concurrent, len(concurrent_calls))
+            time.sleep(0.5)
+            concurrent_calls.pop()
+
+        mock_exporter.upsert_span = Mock(side_effect=slow_upsert)
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=3)
+
+        # Submit 10 tasks rapidly
+        spans = [
+            self.create_mock_span({"span_type": "eval", "id": str(i)})
+            for i in range(10)
+        ]
+
+        for span in spans:
+            processor.on_start(span, None)
+
+        # Wait for all to complete
+        time.sleep(2)
+
+        # Max concurrent should not exceed max_workers (3)
+        assert max_concurrent <= 3, (
+            f"Max concurrent was {max_concurrent}, expected <= 3"
+        )
+
+    def test_shutdown_waits_for_pending_tasks(self, mock_exporter):
+        """Test that shutdown properly cleans up the thread pool."""
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=2)
+
+        # Submit some tasks
+        for i in range(3):
+            span = self.create_mock_span({"span_type": "eval", "id": str(i)})
+            processor.on_start(span, None)
+
+        # Shutdown should complete without errors
+        processor.shutdown()
+
+        # Verify executor is shutdown (calling shutdown multiple times should be safe)
+        processor.shutdown()  # Should not raise
+
+    def test_multiple_processors_independent_thread_pools(self, mock_exporter):
+        """Test that multiple processors have independent thread pools."""
+        processor1 = LiveTrackingSpanProcessor(mock_exporter, max_workers=5)
+        processor2 = LiveTrackingSpanProcessor(mock_exporter, max_workers=15)
+
+        assert processor1.executor != processor2.executor
+        assert processor1.executor._max_workers == 5
+        assert processor2.executor._max_workers == 15
+
+    def test_thread_pool_name_prefix(self, mock_exporter):
+        """Test that thread pool uses correct name prefix."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        # ThreadPoolExecutor sets _thread_name_prefix
+        assert processor.executor._thread_name_prefix == "span-upsert"
+
+    def test_resource_exhaustion_prevention(self, mock_exporter):
+        """Test that max_workers prevents resource exhaustion."""
+        call_times = []
+
+        def timed_upsert(*args, **kwargs):
+            call_times.append(time.time())
+            time.sleep(0.3)
+
+        mock_exporter.upsert_span = Mock(side_effect=timed_upsert)
+        # Very low max_workers to test queueing
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=2)
+
+        # Submit 6 tasks
+        for i in range(6):
+            span = self.create_mock_span({"span_type": "eval", "id": str(i)})
+            processor.on_start(span, None)
+
+        # Wait for all to complete
+        time.sleep(2)
+
+        # All 6 should complete
+        assert len(call_times) == 6
+
+        # With max_workers=2 and 0.3s per task, we should see batching
+        # Sort by time to analyze execution pattern
+        call_times.sort()
+        # First 2 should start quickly, next batch should wait
+        assert call_times[1] - call_times[0] < 0.2  # First batch starts together
+        assert (
+            call_times[3] - call_times[1] > 0.2
+        )  # Second batch waits for first to finish
+
+    def test_shutdown_can_be_called_multiple_times(self, mock_exporter):
+        """Test that shutdown can be safely called multiple times."""
+        processor = LiveTrackingSpanProcessor(mock_exporter)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        processor.on_start(span, None)
+        time.sleep(0.1)
+
+        # Multiple shutdowns should not raise exceptions
+        processor.shutdown()
+        processor.shutdown()
+        processor.shutdown()
+
+    def test_executor_properly_initialized(self, mock_exporter):
+        """Test that ThreadPoolExecutor is properly initialized."""
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=7)
+
+        assert processor.executor is not None
+        assert hasattr(processor.executor, "submit")
+        assert hasattr(processor.executor, "shutdown")
+        assert processor.executor._max_workers == 7