From a8c526b5d0a096ab578219e4f76fe145fc5df42c Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Mon, 7 Sep 2026 11:14:57 +0200 Subject: [PATCH 1/2] feat: stream pooled job logs to the handler over uipath-ipc The pooled `uipath server` now grabs the handler's IIpcLogSink callback during Register (via the injected Message) and points uipath-runtime's process-global log sink at it, so a pooled job's logs stream back over the existing pipe. Best-effort and version-guarded: an older uipath-runtime without the pooled sink API is a no-op and jobs keep the file path. Register(self, message) gains the reach-back handle; the sink forwards each SendLog onto the server loop from the job's worker thread. Bump 2.14.10 -> 2.14.11. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/uipath/pyproject.toml | 2 +- .../uipath/src/uipath/_cli/cli_server_ipc.py | 84 +++++++++++++++- packages/uipath/tests/cli/test_server_ipc.py | 96 ++++++++++++++++++- 3 files changed, 175 insertions(+), 7 deletions(-) diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index 9305fe497..f5ca8f457 100644 --- a/packages/uipath/pyproject.toml +++ b/packages/uipath/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath" -version = "2.14.10" +version = "2.14.11" 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/src/uipath/_cli/cli_server_ipc.py b/packages/uipath/src/uipath/_cli/cli_server_ipc.py index d099f1e23..3342a4a30 100644 --- a/packages/uipath/src/uipath/_cli/cli_server_ipc.py +++ b/packages/uipath/src/uipath/_cli/cli_server_ipc.py @@ -1,9 +1,23 @@ +import asyncio from abc import ABC, abstractmethod +from concurrent.futures import Future from dataclasses import dataclass, field +from typing import TYPE_CHECKING from ._server_core import COMMANDS, _run_command_isolated, _state, parse_args from ._utils._console import ConsoleLogger +if TYPE_CHECKING: + from uipath_ipc import Message +else: + # Optional dependency: only present when the uipath-ipc channel is served (uipath[ipc]). + # The Register annotation stays a string forward-ref so this placeholder is never subscripted + # at import; it is resolved (to the real Message) only during dispatch, which needs uipath-ipc. + try: + from uipath_ipc import Message + except ImportError: # pragma: no cover - no IPC means Register is never dispatched + Message = object + console = ConsoleLogger() @@ -42,8 +56,14 @@ class IPythonRuntimeServer(ABC): """Contract the job executor calls over uipath-ipc.""" @abstractmethod - async def Register(self) -> bool: - """Prove the connection is up. No-op until there is something to register.""" + async def Register(self, message: "Message[None]") -> bool: + """Prove the connection is up, and grab the caller's log-sink callback. + + The injected ``message`` is a reach-back handle (it carries no wire + argument): ``message.client.get_callback`` reaches the handler's + ``IIpcLogSink`` over this same pipe, which pooled jobs stream their logs + into. + """ @abstractmethod async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobResult: @@ -57,7 +77,8 @@ async def StopJob(self, request: PythonServerStopJobRequest) -> bool: class PythonRuntimeService(IPythonRuntimeServer): """``IPythonRuntimeServer`` implementation backed by run/debug/eval.""" - async def Register(self) -> bool: + async def Register(self, message: "Message[None]") -> bool: + _wire_pooled_log_sink(message) console.info("Runtime client registered.") return True @@ -96,6 +117,49 @@ async def StopJob(self, request: PythonServerStopJobRequest) -> bool: return True +def _drain(future: "Future[object]") -> None: + # Retrieve (and discard) a forwarded log's result so a failed one-way send never surfaces. + try: + future.exception() + except BaseException: + pass + + +def _wire_pooled_log_sink(message: "Message[None]") -> None: + """Point the runtime's process-global log sink at this connection's IIpcLogSink callback. + + Pooled jobs run in this same process (``_run_command_isolated`` → a worker thread), so their + logging can reach back to the handler over the pipe the handler dialed in on. The runtime raises + each record to a process-global sink; here we make that sink forward to the handler's callback. + + Best-effort: log streaming is an add-on to the Register handshake, so any failure (an older + uipath-runtime without the pooled sink API, or a peer hosting no callback) leaves jobs on their + file+watcher path and never fails registration. + """ + # A Message built without a caller handle has nothing to reach back to. + client = message.client + if client is None: + return + try: + from uipath.runtime.jobapi import ( # type: ignore[import-untyped] + IIpcLogSink, + set_pooled_log_sink, + ) + except ImportError: + return # runtime predates the pooled sink — nothing to wire; jobs keep the file path + + sink_proxy = client.get_callback(IIpcLogSink) + loop = asyncio.get_running_loop() + + def _forward(job_id: str, log: object) -> None: + # Runs on the job's worker thread: hand the one-way SendLog to the server loop and return at + # once. A dropped log (pipe down) must never surface into the job, so failures are swallowed. + future = asyncio.run_coroutine_threadsafe(sink_proxy.SendLog(job_id, log), loop) + future.add_done_callback(_drain) + + set_pooled_log_sink(_forward) + + async def start_ipc_server(pipe_name: str) -> None: """Serve the Python runtime over a uipath-ipc named pipe until it is closed.""" try: @@ -114,5 +178,15 @@ async def start_ipc_server(pipe_name: str) -> None: request_timeout=None, # jobs are long-running; no server-side timeout ) console.success(f"IPC server listening on pipe '{pipe_name}'") - async with server: - await server.serve_forever() + try: + async with server: + await server.serve_forever() + finally: + # Drop the process-global sink so it can't outlive this loop (matters if the server is + # ever restarted in-process, e.g. in tests). + try: + from uipath.runtime.jobapi import set_pooled_log_sink + + set_pooled_log_sink(None) + except ImportError: + pass diff --git a/packages/uipath/tests/cli/test_server_ipc.py b/packages/uipath/tests/cli/test_server_ipc.py index 801920dd2..3f5a3a708 100644 --- a/packages/uipath/tests/cli/test_server_ipc.py +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -17,6 +17,7 @@ import sys import threading import time +import types from typing import Any, Awaitable, Callable import click @@ -24,6 +25,7 @@ from uipath_ipc import ( IpcClient, IpcServer, + Message, NamedPipeClientTransport, NamedPipeServerTransport, ) @@ -31,6 +33,7 @@ from uipath._cli import _server_core from uipath._cli.cli_server import ( IPythonRuntimeServer, + PythonRuntimeService, PythonServerRunJobResult, start_ipc_server, ) @@ -284,7 +287,7 @@ def test_all_wire_fields_arrive_intact(self): received: list[Any] = [] class SpyService(IPythonRuntimeServer): - async def Register(self) -> bool: + async def Register(self, message: Any) -> bool: return True async def RunJob(self, request: Any) -> PythonServerRunJobResult: @@ -329,3 +332,94 @@ async def drive(proxy: Any) -> None: assert stop_request.JobKey == job_key assert stop_request.ResumeVersion == 5 assert stop_request.ForceStop is True + + +class TestPooledLogSink: + """``Register`` wires the runtime's process-global log sink to the caller's callback. + + This is the pooled path: the handler dials in and hosts an ``IIpcLogSink`` callback; the server + grabs it off the injected ``Message`` and points the runtime's sink at it, so a pooled job's logs + reach back over the same pipe. The runtime side is exercised in uipath-runtime's own tests; here + we lock in the seam (get_callback → set_pooled_log_sink → forward on the loop). + """ + + @staticmethod + def _fake_runtime_jobapi(monkeypatch: Any) -> dict[str, Any]: + """Install a stand-in ``uipath.runtime.jobapi`` and record what the server wires into it.""" + captured: dict[str, Any] = {} + + class IIpcLogSink: # matches the contract the server asks get_callback for + pass + + def set_pooled_log_sink(sink: Any) -> None: + captured["sink"] = sink + + module = types.ModuleType("uipath.runtime.jobapi") + module.IIpcLogSink = IIpcLogSink # type: ignore[attr-defined] + module.set_pooled_log_sink = set_pooled_log_sink # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "uipath.runtime.jobapi", module) + captured["IIpcLogSink"] = IIpcLogSink + return captured + + def test_register_forwards_logs_to_the_caller_callback(self, monkeypatch): + captured = self._fake_runtime_jobapi(monkeypatch) + sent: list[tuple[str, Any]] = [] + + class _Callback: + async def SendLog(self, job_id: str, log: Any) -> None: + sent.append((job_id, log)) + + class _Client: + def __init__(self) -> None: + self.asked_for: Any = None + + def get_callback(self, contract: Any) -> Any: + self.asked_for = contract + return _Callback() + + client = _Client() + + async def scenario() -> None: + ok = await PythonRuntimeService().Register(Message(client=client)) + assert ok is True + # It asked the caller for exactly the IIpcLogSink callback... + assert client.asked_for is captured["IIpcLogSink"] + # ...and registered a forwarder. Driving it (as the runtime would, off a worker thread) + # schedules SendLog on this loop, tagged with the job id. + sink = captured["sink"] + log = {"Message": "hello", "LogLevel": 2} + sink("job-key-42", log) + await asyncio.sleep(0.05) + assert sent == [("job-key-42", log)] + + asyncio.run(scenario()) + + def test_register_is_graceful_when_runtime_lacks_pooled_sink(self, monkeypatch): + # An older uipath-runtime has no jobapi module: importing it raises, and Register must still + # succeed (the job simply keeps its file+watcher log path). + monkeypatch.setitem(sys.modules, "uipath.runtime.jobapi", None) + + class _Client: + def get_callback( + self, contract: Any + ) -> Any: # pragma: no cover - never reached + raise AssertionError( + "must not reach get_callback without the runtime API" + ) + + async def scenario() -> None: + assert ( + await PythonRuntimeService().Register(Message(client=_Client())) is True + ) + + asyncio.run(scenario()) + + def test_register_without_a_caller_handle_is_a_noop(self, monkeypatch): + # A Message with no client (defensive; real dispatch always injects one) wires nothing. + captured = self._fake_runtime_jobapi(monkeypatch) + + async def scenario() -> None: + assert await PythonRuntimeService().Register(Message()) is True + + asyncio.run(scenario()) + assert "sink" not in captured From bea284fc726e3736680962897d38b526aa333096 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Tue, 8 Sep 2026 14:02:53 +0200 Subject: [PATCH 2/2] feat(cli): stream job logs and result to the handler over uipath-ipc uipath-python owns the uipath-ipc contract and connection (_job_api.py) and points uipath-runtime's in-memory sinks at it. The contract is IJobInvocationCommonApi (SendLog + SetResult), the runtime-agnostic base the handler hosts; the JS-only telemetry channel lives on a .NET superset Python doesn't use, so Python keys on the common base directly. Pooled: `uipath server` grabs the handler's IJobInvocationCommonApi callback at Register and installs the sinks per job. Non-pooled: `uipath run --handler-ipc-pipe` dials the handler's per-job server. The result envelope travels inline; output arguments go off-heap via a file pointer. Review fixes: - run the non-pooled handler connection on its own loop/thread so the result sink's blocking ack can't deadlock the job's own event loop - install/clear the process-global sinks inside the job core's lock so concurrently-dispatched RunJobs can't corrupt each other's routing - log (not silently swallow) an IPC result-delivery failure - always disconnect the handler IPC on the run's exit path (async context manager) - register the default runtime factory in start_ipc_server - pin the DTO wire key sets and the log-level mapping in tests Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/uipath/src/uipath/_cli/_job_api.py | 289 ++++++++++++++++++ .../uipath/src/uipath/_cli/_server_core.py | 22 +- packages/uipath/src/uipath/_cli/cli_run.py | 17 +- .../uipath/src/uipath/_cli/cli_server_ipc.py | 116 ++++--- packages/uipath/tests/cli/test_job_api.py | 284 +++++++++++++++++ packages/uipath/tests/cli/test_server_ipc.py | 150 +++++---- packages/uipath/uv.lock | 4 +- 7 files changed, 743 insertions(+), 139 deletions(-) create mode 100644 packages/uipath/src/uipath/_cli/_job_api.py create mode 100644 packages/uipath/tests/cli/test_job_api.py diff --git a/packages/uipath/src/uipath/_cli/_job_api.py b/packages/uipath/src/uipath/_cli/_job_api.py new file mode 100644 index 000000000..21666c1f7 --- /dev/null +++ b/packages/uipath/src/uipath/_cli/_job_api.py @@ -0,0 +1,289 @@ +"""The job-invocation IPC contract and the glue that routes a job's logs and result over it.""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import sys +import threading +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator +from concurrent.futures import Future +from dataclasses import dataclass +from enum import IntEnum +from typing import Any + +logger = logging.getLogger(__name__) + +_SET_RESULT_TIMEOUT_S = 30.0 + + +class LogLevel(IntEnum): + """Log-level wire values.""" + + TRACE = 0 + DEBUG = 1 + INFORMATION = 2 + WARNING = 3 + ERROR = 4 + CRITICAL = 5 + NONE = 6 + + +class ExecutorJobStatus(IntEnum): + """Job-status wire values.""" + + RUNNING = 1 + FAULTED = 2 + SUCCESSFUL = 3 + STOPPED = 4 + SUSPENDED = 5 + + +@dataclass +class JobLogDto: + """A log entry; field names are the wire keys (do not rename).""" + + Message: str = "" + LogLevel: int = LogLevel.INFORMATION.value + + +@dataclass +class JobExecutorError: + """A result error; field names are the wire keys (do not rename).""" + + Code: str | None = None + Title: str | None = None + Detail: str | None = None + Category: str | None = None + Status: int | None = None + + +@dataclass +class JobResultDto: + """The final result; field names are the wire keys (do not rename).""" + + id: str = "" + status: int = ExecutorJobStatus.SUCCESSFUL.value + outputArguments: Any = None + outputArgumentsFilePath: str | None = None + info: str | None = None + error: JobExecutorError | None = None + + +class IJobInvocationCommonApi(ABC): + """The job-invocation contract: logs + the final result. The class name is the endpoint key.""" + + @abstractmethod + async def SendLog(self, jobId: str, log: JobLogDto) -> None: + """Forward one log entry.""" + + @abstractmethod + async def SetResult(self, jobId: str, result: JobResultDto) -> bool: + """Submit the final result.""" + + +def _to_log_level(levelno: int) -> int: + if levelno >= logging.CRITICAL: + return LogLevel.CRITICAL + if levelno >= logging.ERROR: + return LogLevel.ERROR + if levelno >= logging.WARNING: + return LogLevel.WARNING + if levelno >= logging.INFO: + return LogLevel.INFORMATION + if levelno >= logging.DEBUG: + return LogLevel.DEBUG + return LogLevel.TRACE + + +_EXECUTOR_STATUS: dict[str, int] = { + "successful": ExecutorJobStatus.SUCCESSFUL.value, + "faulted": ExecutorJobStatus.FAULTED.value, + "suspended": ExecutorJobStatus.SUSPENDED.value, +} + + +def _to_result_dto( + job_id: str, result: Any, output_arguments_file_path: str +) -> JobResultDto: + error = None + if result is not None and getattr(result, "error", None) is not None: + category = result.error.category + error = JobExecutorError( + Code=result.error.code, + Title=result.error.title, + Detail=result.error.detail, + Category=getattr(category, "value", category), + Status=result.error.status, + ) + raw_status = getattr(result, "status", None) + status_key = str(getattr(raw_status, "value", raw_status) or "successful").lower() + return JobResultDto( + id=job_id, + status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value), + outputArgumentsFilePath=output_arguments_file_path, + error=error, + ) + + +def _drain(future: "Future[object]") -> None: + try: + future.exception() + except BaseException: + pass + + +class _IpcLogHandler(logging.Handler): + """Forwards each log record to the callback.""" + + def __init__( + self, job_id: str, callback: Any, loop: asyncio.AbstractEventLoop + ) -> None: + super().__init__() + self._job_id = job_id + self._callback = callback + self._loop = loop + + def emit(self, record: logging.LogRecord) -> None: + try: + message = self.format(record) + dto = JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno)) + future = asyncio.run_coroutine_threadsafe( + self._callback.SendLog(self._job_id, dto), self._loop + ) + future.add_done_callback(_drain) + except Exception: + self.handleError(record) + + +def install_runtime_sinks( + job_id: str, callback: Any, loop: asyncio.AbstractEventLoop +) -> None: + """Install the log + result sinks, forwarding to ``callback`` on ``loop``. + + ``loop`` must run on a different thread than the one the sinks are invoked on, or the result ack + deadlocks. No-op if the sinks aren't available. + """ + try: + from uipath.runtime.output_sinks import ( # type: ignore[import-untyped] + set_log_handler, + set_result_sink, + ) + except ImportError: + return + + handler = _IpcLogHandler(job_id, callback, loop) + handler.setFormatter(logging.Formatter("%(message)s")) + + def _result_sink(result: Any, output_arguments_file_path: str) -> None: + dto = _to_result_dto(job_id, result, output_arguments_file_path) + try: + future = asyncio.run_coroutine_threadsafe( + callback.SetResult(job_id, dto), loop + ) + future.result(timeout=_SET_RESULT_TIMEOUT_S) + except Exception: + # Best-effort — a dropped result must be logged, not swallowed. + logger.exception("Failed to deliver job result over IPC (SetResult)") + + set_log_handler(handler) + set_result_sink(_result_sink) + + +def clear_runtime_sinks() -> None: + """Clear the installed sinks.""" + try: + from uipath.runtime.output_sinks import set_log_handler, set_result_sink + except ImportError: + return + set_log_handler(None) + set_result_sink(None) + + +def _new_ipc_event_loop() -> asyncio.AbstractEventLoop: + """A fresh event loop for the connection's thread (Proactor on Windows, required for named pipes).""" + if sys.platform == "win32": + return asyncio.ProactorEventLoop() + return asyncio.new_event_loop() + + +class _HandlerIpcConnection: + """An IPC client on its own loop/thread, so the synchronous result ack can't deadlock the job's loop.""" + + def __init__( + self, + client: Any, + loop: asyncio.AbstractEventLoop, + thread: threading.Thread, + ) -> None: + self._client = client + self._loop = loop + self._thread = thread + + def _shutdown(self) -> None: + """Close the client and stop its loop/thread.""" + try: + asyncio.run_coroutine_threadsafe(self._client.aclose(), self._loop).result( + timeout=_SET_RESULT_TIMEOUT_S + ) + except Exception: + pass + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=_SET_RESULT_TIMEOUT_S) + self._loop.close() + + +def connect_handler_ipc(pipe: str, job_id: str) -> _HandlerIpcConnection: + """Dial ``pipe`` on a dedicated loop/thread and install the sinks (see ``_HandlerIpcConnection``).""" + try: + from uipath_ipc import IpcClient, NamedPipeClientTransport + except ImportError as e: + raise RuntimeError( + "--handler-ipc-pipe requires the 'uipath-ipc' package. Install it (pip install 'uipath[ipc]')." + ) from e + + loop = _new_ipc_event_loop() + thread = threading.Thread( + target=loop.run_forever, name="uipath-handler-ipc", daemon=True + ) + thread.start() + + async def _build() -> Any: + client = IpcClient(transport=NamedPipeClientTransport(pipe)) + proxy = client.get_proxy(IJobInvocationCommonApi) # type: ignore[type-abstract] + return client, proxy + + try: + client, proxy = asyncio.run_coroutine_threadsafe(_build(), loop).result( + timeout=_SET_RESULT_TIMEOUT_S + ) + except BaseException: + # On failure, don't leak the loop/thread. + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=_SET_RESULT_TIMEOUT_S) + loop.close() + raise + # Install in the caller's context, not on the IPC thread: the sinks are contextvars, resolved in + # the job's own context at teardown. The ack still runs on `loop` (a separate thread) — no deadlock. + install_runtime_sinks(job_id, proxy, loop) + return _HandlerIpcConnection(client, loop, thread) + + +async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None: + """Clear the sinks and tear down the connection.""" + clear_runtime_sinks() + # Off the caller's loop so joining the thread doesn't block it. + await asyncio.to_thread(conn._shutdown) + + +@contextlib.asynccontextmanager +async def handler_ipc_connection(pipe: str | None, job_id: str) -> AsyncIterator[Any]: + """Connect (if ``pipe`` is set) and always disconnect on exit; yields the connection or None.""" + conn = connect_handler_ipc(pipe, job_id) if pipe else None + try: + yield conn + finally: + if conn is not None: + await disconnect_handler_ipc(conn) diff --git a/packages/uipath/src/uipath/_cli/_server_core.py b/packages/uipath/src/uipath/_cli/_server_core.py index c426126ff..25d9614c5 100644 --- a/packages/uipath/src/uipath/_cli/_server_core.py +++ b/packages/uipath/src/uipath/_cli/_server_core.py @@ -3,7 +3,7 @@ import asyncio import os import shlex -from typing import Any +from typing import Any, Callable from .cli_debug import debug from .cli_eval import eval @@ -50,8 +50,14 @@ async def _run_command_isolated( args: list[str], env_vars: dict[str, str], working_dir: str | None, + on_run_start: Callable[[], None] | None = None, + on_run_end: Callable[[], None] | None = None, ) -> dict[str, Any]: - """Run one command with per-job env/cwd isolation (the shared job core).""" + """Run one command with per-job env/cwd isolation (the shared job core). + + ``on_run_start`` / ``on_run_end`` run INSIDE the serialization lock, so any per-job process-global + state is visible only while this job runs. + """ if _state.lock is None or _state.baseline_env is None: raise RuntimeError("Server state not initialized") @@ -79,9 +85,15 @@ async def _run_command_isolated( "ClientError": True, } - result_value = await asyncio.to_thread( - cmd.main, args, standalone_mode=False - ) + if on_run_start is not None: + on_run_start() + try: + result_value = await asyncio.to_thread( + cmd.main, args, standalone_mode=False + ) + finally: + if on_run_end is not None: + on_run_end() return { "ExitCode": 0, "Error": None, diff --git a/packages/uipath/src/uipath/_cli/cli_run.py b/packages/uipath/src/uipath/_cli/cli_run.py index 9d12a86c3..db5dcf8a3 100644 --- a/packages/uipath/src/uipath/_cli/cli_run.py +++ b/packages/uipath/src/uipath/_cli/cli_run.py @@ -115,6 +115,12 @@ def get_usage_help(self) -> list[str]: default=None, help="Simulation config as a JSON object (same schema as simulation.json)", ) +@click.option( + "--handler-ipc-pipe", + required=False, + default=None, + help="Named pipe to stream this job's logs and result over uipath-ipc instead of writing them to files.", +) @track_command("run") def run( entrypoint: str | None, @@ -129,6 +135,7 @@ def run( debug_port: int, keep_state_file: bool, simulation: str | None, + handler_ipc_pipe: str | None, ) -> None: """Execute the project.""" input_file = file or input_file @@ -212,8 +219,14 @@ async def execute() -> None: JsonLinesFileExporter(ctx.trace_file) ) - async with ResourceOverwritesContext( - lambda: read_resource_overwrites_from_file(ctx.runtime_dir) + # If a pipe was given, install the sinks around the run (always torn down); else a no-op. + from ._job_api import handler_ipc_connection + + async with ( + handler_ipc_connection(handler_ipc_pipe, ctx.job_id or ""), + ResourceOverwritesContext( + lambda: read_resource_overwrites_from_file(ctx.runtime_dir) + ), ): with ExecutionSourceContext(ctx.execution_source), ctx: base_runtime: UiPathRuntimeProtocol | None = None diff --git a/packages/uipath/src/uipath/_cli/cli_server_ipc.py b/packages/uipath/src/uipath/_cli/cli_server_ipc.py index 3342a4a30..8e0442200 100644 --- a/packages/uipath/src/uipath/_cli/cli_server_ipc.py +++ b/packages/uipath/src/uipath/_cli/cli_server_ipc.py @@ -1,8 +1,7 @@ import asyncio from abc import ABC, abstractmethod -from concurrent.futures import Future from dataclasses import dataclass, field -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from ._server_core import COMMANDS, _run_command_isolated, _state, parse_args from ._utils._console import ConsoleLogger @@ -10,9 +9,8 @@ if TYPE_CHECKING: from uipath_ipc import Message else: - # Optional dependency: only present when the uipath-ipc channel is served (uipath[ipc]). - # The Register annotation stays a string forward-ref so this placeholder is never subscripted - # at import; it is resolved (to the real Message) only during dispatch, which needs uipath-ipc. + # Optional dependency (uipath[ipc]): the Register annotation stays a string forward-ref so this + # placeholder is never subscripted at import — resolved to the real Message only at dispatch. try: from uipath_ipc import Message except ImportError: # pragma: no cover - no IPC means Register is never dispatched @@ -37,6 +35,8 @@ class PythonServerRunRequest: Args: str | list[str] | None = None WorkingDirectory: str | None = None EnvironmentVariables: dict[str, str] = field(default_factory=dict) + # Per-job opt-in (default False so it stays off unless explicitly set). + StreamOutputOverIpc: bool = False @dataclass @@ -57,13 +57,7 @@ class IPythonRuntimeServer(ABC): @abstractmethod async def Register(self, message: "Message[None]") -> bool: - """Prove the connection is up, and grab the caller's log-sink callback. - - The injected ``message`` is a reach-back handle (it carries no wire - argument): ``message.client.get_callback`` reaches the handler's - ``IIpcLogSink`` over this same pipe, which pooled jobs stream their logs - into. - """ + """Prove the connection is up and grab the caller's callback via ``message.client.get_callback``.""" @abstractmethod async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobResult: @@ -77,8 +71,23 @@ async def StopJob(self, request: PythonServerStopJobRequest) -> bool: class PythonRuntimeService(IPythonRuntimeServer): """``IPythonRuntimeServer`` implementation backed by run/debug/eval.""" + def __init__(self) -> None: + # The caller's callback, grabbed at Register (None until then). + self._callback: Any = None + self._loop: "asyncio.AbstractEventLoop | None" = None + async def Register(self, message: "Message[None]") -> bool: - _wire_pooled_log_sink(message) + client = message.client + if client is not None: + try: + from ._job_api import IJobInvocationCommonApi + + self._callback = client.get_callback(IJobInvocationCommonApi) # type: ignore[type-abstract] + self._loop = asyncio.get_running_loop() + except Exception: + self._callback = ( + None # older runtime / no callback: jobs keep the file path + ) console.info("Runtime client registered.") return True @@ -101,9 +110,27 @@ async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobRes f"Running job {_run_id(request.JobKey, request.ResumeVersion)}: {command_name} {args}" ) + # Only when opted in and a callback exists. Install/clear the sinks INSIDE the job core's lock + # (via the hooks) so they're bound only while THIS job runs. + callback, loop = self._callback, self._loop + on_run_start: "Any" = None + on_run_end: "Any" = None + if request.StreamOutputOverIpc and callback is not None and loop is not None: + from ._job_api import clear_runtime_sinks, install_runtime_sinks + + job_key = request.JobKey + on_run_start = lambda: install_runtime_sinks(job_key, callback, loop) # noqa: E731 + on_run_end = clear_runtime_sinks + result = await _run_command_isolated( - cmd, args, request.EnvironmentVariables, request.WorkingDirectory + cmd, + args, + request.EnvironmentVariables, + request.WorkingDirectory, + on_run_start=on_run_start, + on_run_end=on_run_end, ) + # IPC contract (PythonServerRunJobResult) carries only ExitCode + Error. return PythonServerRunJobResult( ExitCode=result["ExitCode"], Error=result["Error"] @@ -117,49 +144,6 @@ async def StopJob(self, request: PythonServerStopJobRequest) -> bool: return True -def _drain(future: "Future[object]") -> None: - # Retrieve (and discard) a forwarded log's result so a failed one-way send never surfaces. - try: - future.exception() - except BaseException: - pass - - -def _wire_pooled_log_sink(message: "Message[None]") -> None: - """Point the runtime's process-global log sink at this connection's IIpcLogSink callback. - - Pooled jobs run in this same process (``_run_command_isolated`` → a worker thread), so their - logging can reach back to the handler over the pipe the handler dialed in on. The runtime raises - each record to a process-global sink; here we make that sink forward to the handler's callback. - - Best-effort: log streaming is an add-on to the Register handshake, so any failure (an older - uipath-runtime without the pooled sink API, or a peer hosting no callback) leaves jobs on their - file+watcher path and never fails registration. - """ - # A Message built without a caller handle has nothing to reach back to. - client = message.client - if client is None: - return - try: - from uipath.runtime.jobapi import ( # type: ignore[import-untyped] - IIpcLogSink, - set_pooled_log_sink, - ) - except ImportError: - return # runtime predates the pooled sink — nothing to wire; jobs keep the file path - - sink_proxy = client.get_callback(IIpcLogSink) - loop = asyncio.get_running_loop() - - def _forward(job_id: str, log: object) -> None: - # Runs on the job's worker thread: hand the one-way SendLog to the server loop and return at - # once. A dropped log (pipe down) must never surface into the job, so failures are swallowed. - future = asyncio.run_coroutine_threadsafe(sink_proxy.SendLog(job_id, log), loop) - future.add_done_callback(_drain) - - set_pooled_log_sink(_forward) - - async def start_ipc_server(pipe_name: str) -> None: """Serve the Python runtime over a uipath-ipc named pipe until it is closed.""" try: @@ -172,6 +156,12 @@ async def start_ipc_server(pipe_name: str) -> None: ) from e _state.init() + + # Register the default runtime factory (idempotent) so the server works when started outside the CLI. + from uipath._cli import _ensure_runtime_initialized + + _ensure_runtime_initialized() + server = IpcServer( transport=NamedPipeServerTransport(pipe_name), services={IPythonRuntimeServer: PythonRuntimeService()}, @@ -182,11 +172,7 @@ async def start_ipc_server(pipe_name: str) -> None: async with server: await server.serve_forever() finally: - # Drop the process-global sink so it can't outlive this loop (matters if the server is - # ever restarted in-process, e.g. in tests). - try: - from uipath.runtime.jobapi import set_pooled_log_sink - - set_pooled_log_sink(None) - except ImportError: - pass + # Drop the sinks so they can't outlive this loop (matters on in-process restart). + from ._job_api import clear_runtime_sinks + + clear_runtime_sinks() diff --git a/packages/uipath/tests/cli/test_job_api.py b/packages/uipath/tests/cli/test_job_api.py new file mode 100644 index 000000000..0317f0337 --- /dev/null +++ b/packages/uipath/tests/cli/test_job_api.py @@ -0,0 +1,284 @@ +"""The uipath-python job-api glue: result mapping and the runtime-sink installer. + +The runtime side (``uipath.runtime.output_sinks``) is faked here so these tests exercise the wiring +in isolation — that install points a log handler + result sink at the callback, that the handler +forwards SendLog, and that the result sink maps the runtime result and calls SetResult. +""" + +import asyncio +import logging +import os +import sys +import threading +import types +from typing import Any + +import pytest + +from uipath._cli import _job_api + + +def test_to_result_dto_maps_status_error_and_path(): + class _Category: + value = "User" + + class _Error: + code = "BOOM" + title = "It broke" + detail = "stack" + category = _Category() + status = None + + class _Result: + status = "faulted" + error = _Error() + + dto = _job_api._to_result_dto("job-1", _Result(), "out.args") + + assert dto.id == "job-1" + assert dto.status == _job_api.ExecutorJobStatus.FAULTED.value + assert dto.outputArgumentsFilePath == "out.args" + assert dto.outputArguments is None + assert dto.error is not None + assert (dto.error.Code, dto.error.Category) == ("BOOM", "User") + + +def test_to_result_dto_defaults_to_successful_without_error(): + class _Result: + status = "successful" + error = None + + dto = _job_api._to_result_dto("j", _Result(), "p.args") + assert dto.status == _job_api.ExecutorJobStatus.SUCCESSFUL.value + assert dto.error is None + + +def test_to_log_level_maps_python_levels_to_wire_values(): + assert _job_api._to_log_level(logging.CRITICAL) == _job_api.LogLevel.CRITICAL + assert _job_api._to_log_level(logging.ERROR) == _job_api.LogLevel.ERROR + assert _job_api._to_log_level(logging.WARNING) == _job_api.LogLevel.WARNING + assert _job_api._to_log_level(logging.INFO) == _job_api.LogLevel.INFORMATION + assert _job_api._to_log_level(logging.DEBUG) == _job_api.LogLevel.DEBUG + assert _job_api._to_log_level(logging.NOTSET) == _job_api.LogLevel.TRACE + # A level between two named severities rounds down to the lower one. + assert _job_api._to_log_level(logging.WARNING + 5) == _job_api.LogLevel.WARNING + + +def test_dto_wire_key_sets_are_pinned(): + """Pin each DTO's on-wire JSON keys so an accidental rename is caught on this side. + + Guards our half of the wire contract: JobResultDto is camelCase, JobLogDto / JobExecutorError + are PascalCase. + """ + serialization = pytest.importorskip("uipath_ipc.wire.serialization") + to_wire = serialization.to_wire + + result_keys = set( + to_wire(_job_api.JobResultDto(id="j", outputArgumentsFilePath="p.args")) + ) + assert result_keys == { + "id", + "status", + "outputArguments", + "outputArgumentsFilePath", + "info", + "error", + } + assert set(to_wire(_job_api.JobLogDto(Message="m"))) == {"Message", "LogLevel"} + assert set(to_wire(_job_api.JobExecutorError(Code="c"))) == { + "Code", + "Title", + "Detail", + "Category", + "Status", + } + + +def _fake_output_sinks(monkeypatch) -> dict[str, Any]: + captured: dict[str, Any] = {} + module = types.ModuleType("uipath.runtime.output_sinks") + module.set_log_handler = lambda h: captured.__setitem__("handler", h) # type: ignore[attr-defined] + module.set_result_sink = lambda s: captured.__setitem__("sink", s) # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "uipath.runtime.output_sinks", module) + return captured + + +def test_install_wires_log_handler_and_result_sink(monkeypatch): + captured = _fake_output_sinks(monkeypatch) + logs: list[tuple[str, Any]] = [] + results: list[tuple[str, Any]] = [] + + class _Callback: + async def SendLog(self, jid: str, dto: Any) -> None: + logs.append((jid, dto)) + + async def SetResult(self, jid: str, dto: Any) -> bool: + results.append((jid, dto)) + return True + + async def scenario() -> None: + loop = asyncio.get_running_loop() + _job_api.install_runtime_sinks("job-7", _Callback(), loop) + + # The log handler forwards each record as SendLog, tagged with the job id. + handler = captured["handler"] + handler.emit( + logging.LogRecord("n", logging.WARNING, "p", 1, "hi %s", ("there",), None) + ) + await asyncio.sleep(0.05) + assert logs[0][0] == "job-7" + assert logs[0][1].Message == "hi there" + assert logs[0][1].LogLevel == _job_api.LogLevel.WARNING.value + + # The result sink maps the result and calls SetResult, off a worker thread, for the ack. + class _Result: + status = "successful" + error = None + + sink = captured["sink"] + await asyncio.to_thread(sink, _Result(), "out.args") + assert results[0][0] == "job-7" + assert results[0][1].outputArgumentsFilePath == "out.args" + + asyncio.run(scenario()) + + +def test_install_is_a_noop_without_the_runtime(monkeypatch): + # An older uipath-runtime has no output_sinks module: install/clear must not raise. + monkeypatch.setitem(sys.modules, "uipath.runtime.output_sinks", None) + _job_api.install_runtime_sinks("j", object(), asyncio.new_event_loop()) + _job_api.clear_runtime_sinks() + + +def test_connect_installs_sinks_and_disconnect_clears(monkeypatch): + pytest.importorskip("uipath_ipc") + captured = _fake_output_sinks(monkeypatch) + + async def scenario() -> None: + # A named-pipe client connects lazily, so no server is needed to build it. + client = _job_api.connect_handler_ipc("some-pipe", "job-1") + assert captured["handler"] is not None + assert captured["sink"] is not None + + await _job_api.disconnect_handler_ipc(client) + assert captured["handler"] is None + assert captured["sink"] is None + + asyncio.run(scenario()) + + +def test_connect_without_uipath_ipc_raises(monkeypatch): + monkeypatch.setitem(sys.modules, "uipath_ipc", None) + with pytest.raises(RuntimeError, match="uipath-ipc"): + _job_api.connect_handler_ipc("pipe", "job-1") + + +_jobapi_pipe_counter = 0 + + +def _unique_jobapi_pipe() -> str: + global _jobapi_pipe_counter + _jobapi_pipe_counter += 1 + return f"uipath-jobapi-test-{os.getpid()}-{_jobapi_pipe_counter}" + + +def _serve_jobapi_in_background(pipe: str, api: Any): + """Host ``api`` as IJobInvocationCommonApi on ``pipe`` in a daemon thread; return a stop() callable. + + The server runs on its OWN loop/thread so it can keep accepting while the test thread is + blocked inside the (synchronous) result sink — the whole point of the regression below. + """ + from uipath_ipc import IpcServer, NamedPipeServerTransport + + loop = asyncio.new_event_loop() + ready = threading.Event() + holder: dict[str, Any] = {} + + async def _serve() -> None: + server = IpcServer( + transport=NamedPipeServerTransport(pipe), + services={_job_api.IJobInvocationCommonApi: api}, + request_timeout=None, + ) + holder["server"] = server + async with server: # __aenter__ binds the listener + ready.set() + await server.serve_forever() + + def _run() -> None: + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(_serve()) + except Exception: + pass + finally: + loop.close() + + thread = threading.Thread(target=_run, name="jobapi-test-server", daemon=True) + thread.start() + if not ready.wait(timeout=10): + raise TimeoutError(f"job-api test server on {pipe!r} did not start") + + def stop() -> None: + server = holder.get("server") + if server is not None: + try: + asyncio.run_coroutine_threadsafe(server.aclose(), loop).result( + timeout=10 + ) + except Exception: + pass + thread.join(timeout=10) + + return stop + + +def test_result_sink_delivers_when_invoked_on_the_caller_loop_thread(monkeypatch): + """Regression for the non-pooled deadlock. + + Under ``uipath run`` the runtime invokes the result sink synchronously on the job's own asyncio + loop thread, and the sink blocks for the handler's ack. If that ack were scheduled onto the same + loop it could never run — 30s timeout, dropped result. ``connect_handler_ipc`` isolates the + connection on its own loop, so the ack still completes. Here we drive the real sink on the + caller's loop thread against a real in-proc server and assert SetResult actually arrived. + """ + pytest.importorskip("uipath_ipc") + captured = _fake_output_sinks(monkeypatch) + received: dict[str, Any] = {} + + class _Api(_job_api.IJobInvocationCommonApi): + async def SendLog(self, jobId: str, log: Any) -> None: + received.setdefault("logs", []).append((jobId, log)) + + async def SetResult(self, jobId: str, result: Any) -> bool: + # The server deserializes against the contract's typed signature, so result arrives as a + # real JobResultDto (this also exercises the wire round-trip of the DTO). + received["result"] = (jobId, result) + return True + + class _Result: + status = "successful" + error = None + + pipe = _unique_jobapi_pipe() + stop = _serve_jobapi_in_background(pipe, _Api()) + try: + + async def scenario() -> None: + conn = _job_api.connect_handler_ipc(pipe, "job-77") + # Call the sink ON this loop's thread, exactly as the runtime's __exit__ does. Before the + # fix this deadlocked the loop the ack was scheduled on; now it completes. + captured["sink"](_Result(), "out.args") + await _job_api.disconnect_handler_ipc(conn) + + asyncio.run(scenario()) + finally: + stop() + + assert "result" in received, ( + "SetResult never arrived — the result sink deadlocked/timed out" + ) + job_id, dto = received["result"] + assert job_id == "job-77" + assert dto.outputArgumentsFilePath == "out.args" + assert dto.status == _job_api.ExecutorJobStatus.SUCCESSFUL.value diff --git a/packages/uipath/tests/cli/test_server_ipc.py b/packages/uipath/tests/cli/test_server_ipc.py index 3f5a3a708..6a6dc25ad 100644 --- a/packages/uipath/tests/cli/test_server_ipc.py +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -17,7 +17,6 @@ import sys import threading import time -import types from typing import Any, Awaitable, Callable import click @@ -334,40 +333,11 @@ async def drive(proxy: Any) -> None: assert stop_request.ForceStop is True -class TestPooledLogSink: - """``Register`` wires the runtime's process-global log sink to the caller's callback. +class TestPooledSinks: + """Register grabs the handler's callback; RunJob installs the runtime sinks around each job.""" - This is the pooled path: the handler dials in and hosts an ``IIpcLogSink`` callback; the server - grabs it off the injected ``Message`` and points the runtime's sink at it, so a pooled job's logs - reach back over the same pipe. The runtime side is exercised in uipath-runtime's own tests; here - we lock in the seam (get_callback → set_pooled_log_sink → forward on the loop). - """ - - @staticmethod - def _fake_runtime_jobapi(monkeypatch: Any) -> dict[str, Any]: - """Install a stand-in ``uipath.runtime.jobapi`` and record what the server wires into it.""" - captured: dict[str, Any] = {} - - class IIpcLogSink: # matches the contract the server asks get_callback for - pass - - def set_pooled_log_sink(sink: Any) -> None: - captured["sink"] = sink - - module = types.ModuleType("uipath.runtime.jobapi") - module.IIpcLogSink = IIpcLogSink # type: ignore[attr-defined] - module.set_pooled_log_sink = set_pooled_log_sink # type: ignore[attr-defined] - monkeypatch.setitem(sys.modules, "uipath.runtime.jobapi", module) - captured["IIpcLogSink"] = IIpcLogSink - return captured - - def test_register_forwards_logs_to_the_caller_callback(self, monkeypatch): - captured = self._fake_runtime_jobapi(monkeypatch) - sent: list[tuple[str, Any]] = [] - - class _Callback: - async def SendLog(self, job_id: str, log: Any) -> None: - sent.append((job_id, log)) + def test_register_grabs_the_callback(self): + from uipath._cli import _job_api class _Client: def __init__(self) -> None: @@ -375,51 +345,101 @@ def __init__(self) -> None: def get_callback(self, contract: Any) -> Any: self.asked_for = contract - return _Callback() + return "CALLBACK" client = _Client() + service = PythonRuntimeService() async def scenario() -> None: - ok = await PythonRuntimeService().Register(Message(client=client)) - assert ok is True - # It asked the caller for exactly the IIpcLogSink callback... - assert client.asked_for is captured["IIpcLogSink"] - # ...and registered a forwarder. Driving it (as the runtime would, off a worker thread) - # schedules SendLog on this loop, tagged with the job id. - sink = captured["sink"] - log = {"Message": "hello", "LogLevel": 2} - sink("job-key-42", log) - await asyncio.sleep(0.05) - assert sent == [("job-key-42", log)] + assert await service.Register(Message(client=client)) is True asyncio.run(scenario()) + assert client.asked_for is _job_api.IJobInvocationCommonApi + assert service._callback == "CALLBACK" + assert service._loop is not None - def test_register_is_graceful_when_runtime_lacks_pooled_sink(self, monkeypatch): - # An older uipath-runtime has no jobapi module: importing it raises, and Register must still - # succeed (the job simply keeps its file+watcher log path). - monkeypatch.setitem(sys.modules, "uipath.runtime.jobapi", None) + def test_register_without_a_client_is_a_noop(self): + service = PythonRuntimeService() - class _Client: - def get_callback( - self, contract: Any - ) -> Any: # pragma: no cover - never reached - raise AssertionError( - "must not reach get_callback without the runtime API" - ) + async def scenario() -> None: + assert await service.Register(Message()) is True + + asyncio.run(scenario()) + assert service._callback is None + + def test_runjob_installs_then_clears_the_sinks(self, monkeypatch): + from uipath._cli import _job_api, cli_server_ipc + from uipath._cli.cli_server_ipc import PythonServerRunRequest + + events: list[tuple[Any, ...]] = [] + monkeypatch.setattr( + _job_api, + "install_runtime_sinks", + lambda jid, cb, loop: events.append(("install", jid, cb)), + ) + monkeypatch.setattr( + _job_api, "clear_runtime_sinks", lambda: events.append(("clear",)) + ) + + async def _fake_run(cmd, args, env, wd, on_run_start=None, on_run_end=None): + # The real core runs the hooks inside its lock, around the job; mirror that here. + if on_run_start: + on_run_start() + events.append(("run",)) + if on_run_end: + on_run_end() + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) + + service = PythonRuntimeService() + service._callback = "CALLBACK" + service._loop = asyncio.new_event_loop() + request = PythonServerRunRequest( + JobKey="job-9", Command="run", Args=[], StreamOutputOverIpc=True + ) async def scenario() -> None: - assert ( - await PythonRuntimeService().Register(Message(client=_Client())) is True - ) + await service.RunJob(request) asyncio.run(scenario()) + service._loop.close() + assert events == [("install", "job-9", "CALLBACK"), ("run",), ("clear",)] + + def test_runjob_skips_sinks_when_not_opted_in(self, monkeypatch): + from uipath._cli import _job_api, cli_server_ipc + from uipath._cli.cli_server_ipc import PythonServerRunRequest + + events: list[tuple[Any, ...]] = [] + monkeypatch.setattr( + _job_api, + "install_runtime_sinks", + lambda jid, cb, loop: events.append(("install",)), + ) + monkeypatch.setattr( + _job_api, "clear_runtime_sinks", lambda: events.append(("clear",)) + ) + + async def _fake_run(cmd, args, env, wd, on_run_start=None, on_run_end=None): + # The real core runs the hooks inside its lock, around the job; mirror that here. + if on_run_start: + on_run_start() + events.append(("run",)) + if on_run_end: + on_run_end() + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) - def test_register_without_a_caller_handle_is_a_noop(self, monkeypatch): - # A Message with no client (defensive; real dispatch always injects one) wires nothing. - captured = self._fake_runtime_jobapi(monkeypatch) + service = PythonRuntimeService() + service._callback = "CALLBACK" + service._loop = asyncio.new_event_loop() + # StreamOutputOverIpc defaults False -> the handler did not opt this job in. + request = PythonServerRunRequest(JobKey="job-9", Command="run", Args=[]) async def scenario() -> None: - assert await PythonRuntimeService().Register(Message()) is True + await service.RunJob(request) asyncio.run(scenario()) - assert "sink" not in captured + service._loop.close() + assert events == [("run",)] # no install / clear diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index 513633a63..2ce38423d 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -3,7 +3,7 @@ revision = 3 requires-python = ">=3.11" [options] -exclude-newer = "2026-08-25T00:14:27.5403279Z" +exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P2D" [options.exclude-newer-package] @@ -2599,7 +2599,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.14.10" +version = "2.14.11" source = { editable = "." } dependencies = [ { name = "applicationinsights" },