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/_job_api.py b/packages/uipath/src/uipath/_cli/_job_api.py new file mode 100644 index 000000000..f23a95ae6 --- /dev/null +++ b/packages/uipath/src/uipath/_cli/_job_api.py @@ -0,0 +1,336 @@ +"""The uipath-ipc job-invocation contract + the glue that routes a job's logs and result over it. + +This lives in uipath-python (which owns the ``uipath-ipc`` dependency and the IPC connection). +uipath-runtime stays transport-agnostic: it just calls the in-memory sinks installed here via +``uipath.runtime.output_sinks``. ``install_runtime_sinks`` points those sinks at an +``IJobInvocationApi`` callback (the handler, reached over the same pipe); ``clear_runtime_sinks`` +restores the default file behaviour. +""" + +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): + """Mirror of Microsoft.Extensions.Logging.LogLevel (the wire values).""" + + TRACE = 0 + DEBUG = 1 + INFORMATION = 2 + WARNING = 3 + ERROR = 4 + CRITICAL = 5 + NONE = 6 + + +class ExecutorJobStatus(IntEnum): + """Mirror of the handler's ExecutorJobStatus (the wire values).""" + + RUNNING = 1 + FAULTED = 2 + SUCCESSFUL = 3 + STOPPED = 4 + SUSPENDED = 5 + + +@dataclass +class JobLogDto: + """A single log entry (PascalCase to match the wire; .NET has no [JsonProperty]).""" + + Message: str = "" + LogLevel: int = LogLevel.INFORMATION.value + + +@dataclass +class JobExecutorError: + """The result's error (PascalCase; the .NET JobExecutorError has no [JsonProperty]).""" + + 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 job result (camelCase to match the .NET [JsonProperty] wire keys). + + ``outputArguments`` (the unbounded customer output) is never sent inline — the runtime spills it + to a file and this carries only ``outputArgumentsFilePath``. The rest is the bounded envelope. + """ + + id: str = "" + status: int = ExecutorJobStatus.SUCCESSFUL.value + outputArguments: Any = None + outputArgumentsFilePath: str | None = None + info: str | None = None + error: JobExecutorError | None = None + + +class IIpcLogSink(ABC): + """The log channel; the class name is the CoreIpc endpoint key for the logs-only base.""" + + @abstractmethod + async def SendLog(self, jobId: str, log: JobLogDto) -> None: + """Forward one log entry (one-way).""" + + +class IJobInvocationApi(IIpcLogSink): + """The full contract the handler hosts: logs (inherited) plus the final result.""" + + @abstractmethod + async def SetResult(self, jobId: str, result: JobResultDto) -> bool: + """Submit the final job result (request-response: the handler acks).""" + + +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: + """Map a uipath-runtime result to the wire DTO (envelope inline, output arguments as a pointer).""" + 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 record to the handler's SendLog over the callback (one-way, non-blocking).""" + + 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: + """Point uipath-runtime's log handler + result sink at ``callback`` (an IJobInvocationApi proxy). + + Logs are forwarded one-way; the result blocks for the handler's ack. Both schedule onto ``loop`` + (the connection's loop), which MUST be running on a different thread than the one the runtime + invokes the sinks on — otherwise the blocking ack would wait on a loop that can never run it + (deadlock). Pooled: the job runs on a worker thread while ``loop`` is the free server loop; + non-pooled: ``loop`` is a dedicated thread (see ``connect_handler_ipc``). No-op if uipath-runtime + lacks the sinks. + """ + 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 delivery: don't fault the job (output.json still holds the result), but a + # dropped IPC result must be observable, not silent. + logger.exception("Failed to deliver job result over IPC (SetResult)") + + set_log_handler(handler) + set_result_sink(_result_sink) + + +def clear_runtime_sinks() -> None: + """Restore uipath-runtime's default file behaviour (execution.log / output.json).""" + 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 handler IPC connection to run on its own thread. + + On Windows the named-pipe client needs the Proactor loop (``create_pipe_connection``); + ``new_event_loop()`` already yields it there, but build it explicitly so a non-default event + loop policy can't hand back a Selector loop that cannot dial a pipe. + """ + if sys.platform == "win32": + return asyncio.ProactorEventLoop() + return asyncio.new_event_loop() + + +class _HandlerIpcConnection: + """A handler IPC client bound to its OWN event loop on its OWN thread. + + ``uipath run`` enters the runtime context inline under ``asyncio.run(...)``, so the runtime + invokes the result sink synchronously on the job's loop thread. The sink blocks for the + handler's ack (``future.result()``); if that ack were scheduled onto the job's own loop it could + never run — deadlock. Running the connection on a dedicated loop/thread keeps the two apart: the + sink blocks the job thread while this loop delivers the call (mirroring the pooled path, where + the job already runs on a worker thread while the server loop is free). + """ + + 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 on its loop, then stop the loop and join its thread (best-effort).""" + 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 the handler's per-job pipe on a dedicated loop/thread and install the runtime sinks. + + Non-pooled counterpart of the pooled callback: ``uipath run`` runs in its own process, so it + dials the handler's per-job server itself (the handler passed the pipe as ``--handler-ipc-pipe``) + rather than reaching back over an existing connection. The connection gets its OWN loop on its + OWN thread so the result-sink ack — which the runtime invokes inline on the job's loop thread — + cannot deadlock the loop it waits on (see ``_HandlerIpcConnection``). + """ + try: + from uipath_ipc import IpcClient, NamedPipeClientTransport + except ImportError as e: + raise RuntimeError( + "The handler asked for the uipath-ipc job-api channel (--handler-ipc-pipe) but the " + "'uipath-ipc' package is not installed. 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(IJobInvocationApi) # type: ignore[type-abstract] + # Point the sinks at THIS dedicated loop, not the caller's — that is the fix: the sink's + # blocking ack now waits on a loop running on another thread, which is free to run it. + install_runtime_sinks(job_id, proxy, loop) + return client + + try: + client = asyncio.run_coroutine_threadsafe(_build(), loop).result( + timeout=_SET_RESULT_TIMEOUT_S + ) + except BaseException: + # Building the client failed; don't leak the loop/thread we just started. + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=_SET_RESULT_TIMEOUT_S) + loop.close() + raise + return _HandlerIpcConnection(client, loop, thread) + + +async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None: + """Clear the runtime sinks and tear down the connection's loop/thread (best-effort).""" + clear_runtime_sinks() + # Tear down off the caller's loop so joining the connection 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 the handler IPC leg for the run (if ``pipe`` is set) and ALWAYS disconnect on exit. + + Yields the connection (or None when no pipe was given). Guarantees the sinks are cleared and the + connection torn down even if the job raises — a bare connect/disconnect pair would skip that on + the exception path. + """ + 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..35527a9ed 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,15 @@ 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, immediately around the job, + so per-job process-global state (e.g. the IPC output sinks) is visible ONLY while this job runs — + a concurrently-dispatched job cannot observe or clear another job's state. + """ if _state.lock is None or _state.baseline_env is None: raise RuntimeError("Server state not initialized") @@ -79,9 +86,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..89c8ad216 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 of the handler's job-api server to dial back and stream this job's logs and result over uipath-ipc (instead of the execution.log / output.json 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,17 @@ async def execute() -> None: JsonLinesFileExporter(ctx.trace_file) ) - async with ResourceOverwritesContext( - lambda: read_resource_overwrites_from_file(ctx.runtime_dir) + # Non-pooled IPC: dial the handler's per-job pipe and install the runtime sinks + # (logs + result) around the run, so its log handler is in place before the context is + # entered and is always torn down after (even on failure). Absent the flag this is a + # no-op and the runtime keeps writing execution.log / output.json as usual. + 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 d099f1e23..36484f5c7 100644 --- a/packages/uipath/src/uipath/_cli/cli_server_ipc.py +++ b/packages/uipath/src/uipath/_cli/cli_server_ipc.py @@ -1,9 +1,22 @@ +import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any 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() @@ -23,6 +36,9 @@ class PythonServerRunRequest: Args: str | list[str] | None = None WorkingDirectory: str | None = None EnvironmentVariables: dict[str, str] = field(default_factory=dict) + # Per-job opt-in from the handler (its FF): route this job's logs + result over the IPC callback + # instead of the files. Default False so an older handler that never sets it keeps the files. + StreamOutputOverIpc: bool = False @dataclass @@ -42,8 +58,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 job-invocation callback. + + The injected ``message`` is a reach-back handle (it carries no wire + argument): ``message.client.get_callback`` reaches the handler's + ``IJobInvocationApi`` over this same pipe, which pooled jobs stream their + logs and result into. + """ @abstractmethod async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobResult: @@ -57,7 +79,24 @@ async def StopJob(self, request: PythonServerStopJobRequest) -> bool: class PythonRuntimeService(IPythonRuntimeServer): """``IPythonRuntimeServer`` implementation backed by run/debug/eval.""" - async def Register(self) -> bool: + def __init__(self) -> None: + # The handler's IJobInvocationApi callback, grabbed at Register: pooled jobs stream their logs + # and result back through it. None until Register (or if the peer hosts no callback). + self._callback: Any = None + self._loop: "asyncio.AbstractEventLoop | None" = None + + async def Register(self, message: "Message[None]") -> bool: + client = message.client + if client is not None: + try: + from ._job_api import IJobInvocationApi + + self._callback = client.get_callback(IJobInvocationApi) # 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 @@ -80,9 +119,30 @@ async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobRes f"Running job {_run_id(request.JobKey, request.ResumeVersion)}: {command_name} {args}" ) + # Route this pooled job's logs + result back to the handler over the callback — only when the + # handler opted this job in (its per-job FF) and we grabbed a callback at Register. Install and + # clear run INSIDE the job core's serialization lock (via the hooks) so the process-global sinks + # are bound only while THIS job runs — a concurrently-dispatched RunJob can't overwrite or clear + # them mid-run. + 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"] @@ -108,11 +168,27 @@ async def start_ipc_server(pipe_name: str) -> None: ) from e _state.init() + + # Register the default runtime factory, exactly as the `uipath server` CLI dispatch does. That + # dispatch normally runs first; this is a cheap, idempotent safety net so the pooled server always + # has a factory even when started outside the CLI (embedded host, tests) — otherwise a job's + # `uipath run` would fail with "No default factory registered". + from uipath._cli import _ensure_runtime_initialized + + _ensure_runtime_initialized() + server = IpcServer( transport=NamedPipeServerTransport(pipe_name), services={IPythonRuntimeServer: PythonRuntimeService()}, 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 runtime sinks so they can't outlive this loop (matters if the server is ever + # restarted in-process, e.g. in tests). + 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..38b6bf9f1 --- /dev/null +++ b/packages/uipath/tests/cli/test_job_api.py @@ -0,0 +1,285 @@ +"""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 Python-side rename is caught. + + A Python unit test can't detect .NET-side [JsonProperty] drift (that needs a shared fixture or an + integration test) — this guards the uipath-python half of the 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 IJobInvocationApi 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.IJobInvocationApi: 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.IJobInvocationApi): + 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 801920dd2..cfd489512 100644 --- a/packages/uipath/tests/cli/test_server_ipc.py +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -24,6 +24,7 @@ from uipath_ipc import ( IpcClient, IpcServer, + Message, NamedPipeClientTransport, NamedPipeServerTransport, ) @@ -31,6 +32,7 @@ from uipath._cli import _server_core from uipath._cli.cli_server import ( IPythonRuntimeServer, + PythonRuntimeService, PythonServerRunJobResult, start_ipc_server, ) @@ -284,7 +286,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 +331,115 @@ async def drive(proxy: Any) -> None: assert stop_request.JobKey == job_key assert stop_request.ResumeVersion == 5 assert stop_request.ForceStop is True + + +class TestPooledSinks: + """Register grabs the handler's callback; RunJob installs the runtime sinks around each job.""" + + def test_register_grabs_the_callback(self): + from uipath._cli import _job_api + + 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() + service = PythonRuntimeService() + + async def scenario() -> None: + assert await service.Register(Message(client=client)) is True + + asyncio.run(scenario()) + assert client.asked_for is _job_api.IJobInvocationApi + assert service._callback == "CALLBACK" + assert service._loop is not None + + def test_register_without_a_client_is_a_noop(self): + service = PythonRuntimeService() + + 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: + 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) + + 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: + await service.RunJob(request) + + asyncio.run(scenario()) + 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" },