From 2066dd709b3ef42009a8fb5fbeb1a63de8a2c259 Mon Sep 17 00:00:00 2001 From: Andrei Ancuta Date: Mon, 6 Jul 2026 15:23:23 +0300 Subject: [PATCH 1/4] fix: shutdown trace manager, not the executor, between jobs --- packages/uipath/src/uipath/_cli/cli_debug.py | 26 ++- packages/uipath/src/uipath/_cli/cli_run.py | 35 ++-- .../integration/test_trace_shutdown_server.py | 155 ++++++++++++++++++ packages/uipath/tests/cli/test_server.py | 91 ++-------- packages/uipath/tests/cli/utils/server.py | 93 +++++++++++ 5 files changed, 286 insertions(+), 114 deletions(-) create mode 100644 packages/uipath/tests/cli/integration/test_trace_shutdown_server.py create mode 100644 packages/uipath/tests/cli/utils/server.py diff --git a/packages/uipath/src/uipath/_cli/cli_debug.py b/packages/uipath/src/uipath/_cli/cli_debug.py index fa7e7a170..7f542e871 100644 --- a/packages/uipath/src/uipath/_cli/cli_debug.py +++ b/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: diff --git a/packages/uipath/src/uipath/_cli/cli_run.py b/packages/uipath/src/uipath/_cli/cli_run.py index 1fca4b8cb..9d12a86c3 100644 --- a/packages/uipath/src/uipath/_cli/cli_run.py +++ b/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()) diff --git a/packages/uipath/tests/cli/integration/test_trace_shutdown_server.py b/packages/uipath/tests/cli/integration/test_trace_shutdown_server.py new file mode 100644 index 000000000..754c2fe10 --- /dev/null +++ b/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 diff --git a/packages/uipath/tests/cli/test_server.py b/packages/uipath/tests/cli/test_server.py index 7e069bd62..c4d73f361 100644 --- a/packages/uipath/tests/cli/test_server.py +++ b/packages/uipath/tests/cli/test_server.py @@ -3,15 +3,18 @@ import os import subprocess import sys -import threading import time -from typing import Any from urllib.request import urlopen import aiohttp import pytest -from uipath._cli.cli_server import start_tcp_server +from tests.cli.utils.server import ( + get_free_port, + start_cli_server_thread, + start_job, + start_job_with_env, +) SERVER_START_TIMEOUT_SECONDS = 20 SERVER_STOP_TIMEOUT_SECONDS = 5 @@ -38,47 +41,6 @@ def create_uipath_json(script_path: str, entrypoint_name: str = "main"): return {"functions": {entrypoint_name: f"{script_path}:main"}} -async def start_job( - port: int, job_key: str, command: str, args: list[str] -) -> dict[str, Any]: - """Start a job on the server.""" - async with aiohttp.ClientSession() as session: - async with session.post( - f"http://127.0.0.1:{port}/jobs/{job_key}/start", - json={"command": command, "args": args}, - ) as response: - return await response.json() - - -async def start_job_with_env( - port: int, - job_key: str, - command: str, - args: list[str], - env_vars: dict[str, str], -) -> dict[str, Any]: - """Start a job on the server with environment variables.""" - async with aiohttp.ClientSession() as session: - async with session.post( - f"http://127.0.0.1:{port}/jobs/{job_key}/start", - json={ - "command": command, - "args": args, - "environmentVariables": env_vars, - }, - ) as response: - return await response.json() - - -def get_free_port() -> int: - """Use a random available port for testing.""" - import socket - - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] - - def read_process_output(process: subprocess.Popen[str]) -> str: if process.stdout is None: return "" @@ -128,29 +90,12 @@ class TestServer: @pytest.fixture def server_port(self): """Use a random available port for testing.""" - import socket - - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] + return get_free_port() @pytest.fixture def server(self, server_port): """Start the server in a background thread.""" - - def run_server(): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - loop.run_until_complete(start_tcp_server("127.0.0.1", server_port)) - except asyncio.CancelledError: - pass - finally: - loop.close() - - thread = threading.Thread(target=run_server, daemon=True) - thread.start() - time.sleep(0.5) + start_cli_server_thread(server_port) yield server_port @@ -381,11 +326,7 @@ class TestServerEnvIsolation: @pytest.fixture def server_port(self): - import socket - - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] + return get_free_port() @pytest.fixture def env_snapshots(self): @@ -405,19 +346,7 @@ def spy_cmd(): original_commands = cli_server.COMMANDS.copy() cli_server.COMMANDS["spy"] = spy_cmd - def run_server(): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - loop.run_until_complete(start_tcp_server("127.0.0.1", server_port)) - except asyncio.CancelledError: - pass - finally: - loop.close() - - thread = threading.Thread(target=run_server, daemon=True) - thread.start() - time.sleep(0.5) + start_cli_server_thread(server_port) yield server_port diff --git a/packages/uipath/tests/cli/utils/server.py b/packages/uipath/tests/cli/utils/server.py new file mode 100644 index 000000000..abf1681f8 --- /dev/null +++ b/packages/uipath/tests/cli/utils/server.py @@ -0,0 +1,93 @@ +import asyncio +import socket +import threading +import time +from typing import Any +from urllib.request import urlopen + +import aiohttp + +from uipath._cli.cli_server import start_tcp_server +from uipath.functions import register_default_runtime_factory + + +def get_free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _wait_for_health(port: int, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + last_error: Exception | None = None + while time.monotonic() < deadline: + try: + with urlopen(f"http://127.0.0.1:{port}/health", timeout=0.5) as response: + if response.status == 200: + return + except Exception as exc: + last_error = exc + time.sleep(0.05) + raise RuntimeError(f"server on port {port} did not become healthy") from last_error + + +def start_cli_server_thread(port: int) -> threading.Thread: + register_default_runtime_factory() + + def run_server() -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(start_tcp_server("127.0.0.1", port)) + except asyncio.CancelledError: + pass + finally: + loop.close() + + thread = threading.Thread(target=run_server, daemon=True) + thread.start() + _wait_for_health(port) + return thread + + +async def start_job( + port: int, + job_key: str, + command: str, + args: list[str], + working_directory: str | None = None, +) -> dict[str, Any]: + payload: dict[str, Any] = {"command": command, "args": args} + if working_directory is not None: + payload["workingDirectory"] = working_directory + + async with aiohttp.ClientSession() as session: + async with session.post( + f"http://127.0.0.1:{port}/jobs/{job_key}/start", + json=payload, + ) as response: + return await response.json() + + +async def start_job_with_env( + port: int, + job_key: str, + command: str, + args: list[str], + env_vars: dict[str, str], + working_directory: str | None = None, +) -> dict[str, Any]: + payload: dict[str, Any] = { + "command": command, + "args": args, + "environmentVariables": env_vars, + } + if working_directory is not None: + payload["workingDirectory"] = working_directory + + async with aiohttp.ClientSession() as session: + async with session.post( + f"http://127.0.0.1:{port}/jobs/{job_key}/start", + json=payload, + ) as response: + return await response.json() From 9b5fdd692753bf4959536a3150df6310618819d1 Mon Sep 17 00:00:00 2001 From: Andrei Ancuta Date: Tue, 7 Jul 2026 18:00:23 +0300 Subject: [PATCH 2/4] fix: close llmops exporter during live tracking shutdown --- .../tracing/_live_tracking_processor.py | 4 ++++ .../src/uipath/tracing/_otel_exporters.py | 4 ++++ .../eval/test_live_tracking_span_processor.py | 19 +++++++++++++++++++ .../tests/tracing/test_otel_exporters.py | 11 +++++++++++ 4 files changed, 38 insertions(+) diff --git a/packages/uipath/src/uipath/tracing/_live_tracking_processor.py b/packages/uipath/src/uipath/tracing/_live_tracking_processor.py index 4b0aa0b92..a338a87ed 100644 --- a/packages/uipath/src/uipath/tracing/_live_tracking_processor.py +++ b/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.""" diff --git a/packages/uipath/src/uipath/tracing/_otel_exporters.py b/packages/uipath/src/uipath/tracing/_otel_exporters.py index f8ec3aace..6796c9629 100644 --- a/packages/uipath/src/uipath/tracing/_otel_exporters.py +++ b/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, diff --git a/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py b/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py index a20c18d9b..c2f6d4d9b 100644 --- a/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py +++ b/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) diff --git a/packages/uipath/tests/tracing/test_otel_exporters.py b/packages/uipath/tests/tracing/test_otel_exporters.py index 939dbb8cd..c8630b539 100644 --- a/packages/uipath/tests/tracing/test_otel_exporters.py +++ b/packages/uipath/tests/tracing/test_otel_exporters.py @@ -161,6 +161,17 @@ def test_force_flush(exporter): assert exporter.force_flush() is True +def test_shutdown_closes_http_client(mock_env_vars): + with patch("uipath.tracing._otel_exporters.httpx.Client") as mock_client_class: + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + exporter = LlmOpsHttpExporter() + exporter.shutdown() + + mock_client.close.assert_called_once_with() + + def test_get_base_url(): """Test _get_base_url method with different environment configurations.""" # Test with environment variable set From d93890eff50fb36a913d066c6d026713ac275ba7 Mon Sep 17 00:00:00 2001 From: Andrei Ancuta Date: Wed, 8 Jul 2026 11:51:52 +0300 Subject: [PATCH 3/4] feat: bump version --- packages/uipath/pyproject.toml | 2 +- packages/uipath/uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index 035dac2d1..2ca26e087 100644 --- a/packages/uipath/pyproject.toml +++ b/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" diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index 070142638..19081c3f3 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2598,7 +2598,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.13.6" +version = "2.13.7" source = { editable = "." } dependencies = [ { name = "applicationinsights" }, From 5bc8486cb7b82ab23f7227610cedf6680765616f Mon Sep 17 00:00:00 2001 From: Andrei Ancuta Date: Wed, 8 Jul 2026 11:51:52 +0300 Subject: [PATCH 4/4] test: cover live tracking exporter shutdown failure --- .../cli/eval/test_live_tracking_span_processor.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py b/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py index c2f6d4d9b..d26e89985 100644 --- a/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py +++ b/packages/uipath/tests/cli/eval/test_live_tracking_span_processor.py @@ -3,7 +3,7 @@ import threading import time from typing import Any -from unittest.mock import Mock +from unittest.mock import Mock, patch import pytest from opentelemetry import context as context_api @@ -367,6 +367,16 @@ def shutdown(): assert calls == ["upsert", "shutdown"] + def test_shutdown_logs_exporter_shutdown_failure(self, mock_exporter): + mock_exporter.shutdown = Mock(side_effect=RuntimeError("close failed")) + processor = LiveTrackingSpanProcessor(mock_exporter) + + with patch("uipath.tracing._live_tracking_processor.logger.debug") as debug: + processor.shutdown() + + mock_exporter.shutdown.assert_called_once_with() + debug.assert_called_once_with("Exporter shutdown failed: close failed") + 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)