fix: shutdown trace manager, not the executor, between jobs

packages/uipath/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.13.6"
+version = "2.13.7"
 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"

packages/uipath/src/uipath/_cli/cli_debug.py

@@ -138,7 +138,6 @@ async def execute_debug_runtime():
                 with ExecutionSourceContext(ctx.execution_source), ctx:
                     factory: UiPathRuntimeFactoryProtocol | None = None
                     governance_bootstrap: GovernanceBootstrap | None = None
-                    live_tracking_processor: LiveTrackingSpanProcessor | None = None
 
                     try:
                         trigger_poll_interval: float = 5.0
@@ -169,12 +168,11 @@ async def execute_debug_runtime():
 
                         if ctx.job_id:
                             if UiPathConfig.is_tracing_enabled:
-                                live_tracking_processor = LiveTrackingSpanProcessor(
-                                    LlmOpsHttpExporter(),
-                                    settings=trace_settings,
-                                )
                                 trace_manager.add_span_processor(
-                                    live_tracking_processor
+                                    LiveTrackingSpanProcessor(
+                                        LlmOpsHttpExporter(),
+                                        settings=trace_settings,
+                                    )
                                 )
                             trigger_poll_interval = (
                                 0.0  # Polling disabled for production jobs
@@ -260,15 +258,13 @@ async def execute_debug_runtime():
                             await execute_debug_runtime()
 
                     finally:
-                        # Drain runtime-scoped sinks before the factory
-                        # shuts down — the factory may own transports they
-                        # use. (The inner runtime already disposed above.)
-                        if live_tracking_processor is not None:
-                            live_tracking_processor.shutdown()
-                        if governance_bootstrap is not None:
-                            governance_bootstrap.dispose()
-                        if factory:
-                            await factory.dispose()
+                        try:
+                            if governance_bootstrap is not None:
+                                governance_bootstrap.dispose()
+                            if factory:
+                                await factory.dispose()
+                        finally:
+                            trace_manager.shutdown()
 
             asyncio.run(execute_debug_runtime())
         except Exception as e:

packages/uipath/src/uipath/_cli/cli_run.py

@@ -221,7 +221,6 @@ async def execute() -> None:
                         chat_runtime: UiPathRuntimeProtocol | None = None
                         factory: UiPathRuntimeFactoryProtocol | None = None
                         governance_bootstrap: GovernanceBootstrap | None = None
-                        live_tracking_processor: LiveTrackingSpanProcessor | None = None
                         try:
                             factory = UiPathRuntimeFactoryRegistry.get(context=ctx)
 
@@ -296,12 +295,11 @@ async def execute() -> None:
 
                             if ctx.job_id:
                                 if UiPathConfig.is_tracing_enabled:
-                                    live_tracking_processor = LiveTrackingSpanProcessor(
-                                        LlmOpsHttpExporter(),
-                                        settings=trace_settings,
-                                    )
                                     trace_manager.add_span_processor(
-                                        live_tracking_processor
+                                        LiveTrackingSpanProcessor(
+                                            LlmOpsHttpExporter(),
+                                            settings=trace_settings,
+                                        )
                                     )
 
                                 if ctx.conversation_id and ctx.exchange_id:
@@ -318,18 +316,19 @@ async def execute() -> None:
                             else:
                                 ctx.result = await debug_runtime(ctx, runtime)
                         finally:
-                            if chat_runtime:
-                                await chat_runtime.dispose()
-                            if runtime is not None and runtime is not base_runtime:
-                                await runtime.dispose()
-                            if base_runtime is not None:
-                                await base_runtime.dispose()
-                            if live_tracking_processor is not None:
-                                live_tracking_processor.shutdown()
-                            if governance_bootstrap is not None:
-                                governance_bootstrap.dispose()
-                            if factory:
-                                await factory.dispose()
+                            try:
+                                if chat_runtime:
+                                    await chat_runtime.dispose()
+                                if runtime is not None and runtime is not base_runtime:
+                                    await runtime.dispose()
+                                if base_runtime is not None:
+                                    await base_runtime.dispose()
+                                if governance_bootstrap is not None:
+                                    governance_bootstrap.dispose()
+                                if factory:
+                                    await factory.dispose()
+                            finally:
+                                trace_manager.shutdown()
 
             asyncio.run(execute())
 

packages/uipath/src/uipath/tracing/_live_tracking_processor.py

@@ -89,6 +89,10 @@ def shutdown(self) -> None:
             self.executor.shutdown(wait=True)
         except Exception as e:
             logger.debug(f"Executor shutdown failed: {e}")
+        try:
+            self.exporter.shutdown()
+        except Exception as e:
+            logger.debug(f"Exporter shutdown failed: {e}")
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         """Force flush - no-op for live tracking."""

packages/uipath/src/uipath/tracing/_otel_exporters.py

@@ -176,6 +176,10 @@ def force_flush(self, timeout_millis: int = 30000) -> bool:
         """Force flush the exporter."""
         return True
 
+    def shutdown(self) -> None:
+        """Close the HTTP client."""
+        self.http_client.close()
+
     def upsert_span(
         self,
         span: ReadableSpan,

packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py

@@ -348,6 +348,25 @@ def test_shutdown_waits_for_pending_tasks(self, mock_exporter):
         # Verify executor is shutdown (calling shutdown multiple times should be safe)
         processor.shutdown()  # Should not raise
 
+    def test_shutdown_closes_exporter_after_pending_tasks(self, mock_exporter):
+        calls = []
+
+        def upsert(*args, **kwargs):
+            calls.append("upsert")
+
+        def shutdown():
+            calls.append("shutdown")
+
+        mock_exporter.upsert_span = Mock(side_effect=upsert)
+        mock_exporter.shutdown = Mock(side_effect=shutdown)
+        processor = LiveTrackingSpanProcessor(mock_exporter, max_workers=1)
+        span = self.create_mock_span({"span_type": "eval"})
+
+        processor.on_start(span, None)
+        processor.shutdown()
+
+        assert calls == ["upsert", "shutdown"]
+
     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)

packages/uipath/tests/cli/integration/test_trace_shutdown_server.py

@@ -0,0 +1,155 @@
+import asyncio
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any, Generator, cast
+
+import pytest
+
+from tests.cli.utils.server import (
+    get_free_port,
+    start_cli_server_thread,
+    start_job_with_env,
+)
+
+
+class TraceServer(ThreadingHTTPServer):
+    posts: list[bytes]
+
+
+class TraceHandler(BaseHTTPRequestHandler):
+    def do_POST(self) -> None:
+        length = int(self.headers.get("content-length", "0"))
+        body = self.rfile.read(length)
+        cast(TraceServer, self.server).posts.append(body)
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(b"OK")
+
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+
+@pytest.fixture
+def trace_stub() -> Generator[tuple[TraceServer, str], None, None]:
+    port = get_free_port()
+    server = TraceServer(("127.0.0.1", port), TraceHandler)
+    server.posts = []
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    try:
+        yield server, f"http://127.0.0.1:{port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+
+
+def write_project(project: Path) -> None:
+    (project / "entrypoint.py").write_text(
+        """from dataclasses import dataclass
+from uipath.tracing import traced
+
+
+@dataclass
+class Input:
+    message: str
+
+
+@dataclass
+class Output:
+    message: str
+
+
+@traced(name="actual-agent-span")
+def main(input: Input) -> Output:
+    return Output(message=f"ok: {input.message}")
+""",
+        encoding="utf-8",
+    )
+    (project / "uipath.json").write_text(
+        json.dumps({"agents": {"main": "entrypoint.py:main"}}, indent=2),
+        encoding="utf-8",
+    )
+
+
+async def run_traced_job(
+    port: int,
+    project: Path,
+    trace_url: str,
+    job_key: str,
+    command: str,
+    extra_args: list[str],
+) -> tuple[dict[str, Any], Path]:
+    input_file = project / f"{job_key}-input.json"
+    output_file = project / f"{job_key}-output.json"
+    input_file.write_text(json.dumps({"message": job_key}), encoding="utf-8")
+
+    response = await start_job_with_env(
+        port,
+        job_key,
+        command,
+        [
+            "main",
+            "--input-file",
+            str(input_file),
+            "--output-file",
+            str(output_file),
+            *extra_args,
+        ],
+        {
+            "UIPATH_ACCESS_TOKEN": "fake-token",
+            "UIPATH_JOB_KEY": job_key,
+            "UIPATH_ORGANIZATION_ID": "org-123",
+            "UIPATH_TENANT_ID": "tenant-123",
+            "UIPATH_TRACE_BASE_URL": trace_url,
+            "UIPATH_TRACING_ENABLED": "true",
+            "LOG_LEVEL": "DEBUG",
+        },
+        working_directory=str(project),
+    )
+    return response, output_file
+
+
+def read_json(path: Path) -> Any:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+@pytest.mark.parametrize(
+    ("command", "extra_args"),
+    [
+        ("run", []),
+        ("debug", ["--attach", "none"]),
+    ],
+)
+def test_server_runs_two_traced_jobs_after_trace_shutdown(
+    tmp_path: Path,
+    trace_stub: tuple[TraceServer, str],
+    command: str,
+    extra_args: list[str],
+) -> None:
+    port = get_free_port()
+    start_cli_server_thread(port)
+
+    trace_server, trace_url = trace_stub
+    project = tmp_path / "project"
+    project.mkdir()
+    write_project(project)
+
+    job1_response, job1_output = asyncio.run(
+        run_traced_job(port, project, trace_url, "job-1", command, extra_args)
+    )
+    job2_response, job2_output = asyncio.run(
+        run_traced_job(port, project, trace_url, "job-2", command, extra_args)
+    )
+
+    assert job1_response["success"] is True, job1_response
+    assert job2_response["success"] is True, job2_response
+    assert read_json(job1_output) == {"message": "ok: job-1"}
+    assert read_json(job2_output) == {"message": "ok: job-2"}
+
+    result = read_json(project / "__uipath" / "output.json")
+    assert result["status"] == "successful"
+    assert "error" not in result
+    assert trace_server.posts