diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index 6122cd06b..e31f9d05c 100644 --- a/packages/uipath/pyproject.toml +++ b/packages/uipath/pyproject.toml @@ -1,12 +1,12 @@ [project] name = "uipath" -version = "2.14.16" +version = "2.14.17" 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" dependencies = [ "uipath-core>=0.5.30, <0.6.0", - "uipath-runtime>=0.13.1, <0.14.0", + "uipath-runtime>=0.13.5, <0.14.0", "uipath-platform>=0.2.27, <0.3.0", "click>=8.3.3, <9.0.0", "httpx>=0.28.1", 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..1db129062 --- /dev/null +++ b/packages/uipath/src/uipath/_cli/_job_api.py @@ -0,0 +1,408 @@ +"""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 +import uuid +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator +from concurrent.futures import Future +from concurrent.futures import wait as _futures_wait +from dataclasses import dataclass +from enum import IntEnum +from typing import Any, TypeGuard + +logger = logging.getLogger(__name__) + +# Generous on purpose, and a ceiling rather than a delay: a healthy peer acks in under a millisecond, +# so these only elapse when something is already wrong. Nothing throttles emit, so a chatty job can +# leave thousands of sends in flight at teardown -- a peer that is merely slow, not dead, still needs +# time to drain them, and the tail is the part of a log worth having. +_SET_RESULT_TIMEOUT_S = 120.0 +_LOG_FLUSH_TIMEOUT_S = 120.0 +_IPC_REQUEST_TIMEOUT_S = 120.0 +# Inbound frame cap. uipath-ipc defaults to 2 MB, but every .NET server on this contract already +# accepts 30 (PythonJobApiServer, IpcServerDefaults), so matching keeps the pair symmetric. +_MAX_MESSAGE_BYTES = 30 * 1024 * 1024 + +_NO_IPC_LOGGERS = ("uipath_ipc", __name__) + + +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, + # Defensive only: the runtime delivers a result for SUCCESSFUL and FAULTED alone, and the peer + # resolves a suspended job from output.json (resume triggers never cross the wire). + "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 _write_original_stderr(text: str) -> None: + stream = sys.__stderr__ + if stream is None: + return + try: + stream.write(text + "\n") + stream.flush() + except Exception: + pass + + +def _to_original_stderr(handler: logging.Handler, record: logging.LogRecord) -> None: + _write_original_stderr(handler.format(record)) + + +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 + self._pending: set[Future[object]] = set() + self._pending_lock = threading.Lock() + self._reported_failure = False + + def emit(self, record: logging.LogRecord) -> None: + if record.name.startswith(_NO_IPC_LOGGERS): + _to_original_stderr(self, record) + return + try: + dto = JobLogDto( + Message=self.format(record), LogLevel=_to_log_level(record.levelno) + ) + future = asyncio.run_coroutine_threadsafe( + self._callback.SendLog(self._job_id, dto), self._loop + ) + with self._pending_lock: + self._pending.add(future) + future.add_done_callback(self._settled) + except Exception: + self.handleError(record) + + def _settled(self, future: "Future[object]") -> None: + with self._pending_lock: + self._pending.discard(future) + try: + error: BaseException | None = future.exception() + except BaseException as e: # cancelled + error = e + if error is None: + return + with self._pending_lock: + first = not self._reported_failure + self._reported_failure = True + if first: + # Once per handler: a broken channel fails every later send too. + _write_original_stderr( + f"uipath: job logs are no longer reaching the handler ({error!r}); " + "further log-delivery errors for this job are suppressed." + ) + + def _snapshot(self) -> "list[Future[object]]": + with self._pending_lock: + return list(self._pending) + + def flush_pending(self, timeout: float = _LOG_FLUSH_TIMEOUT_S) -> None: + # Blocks; only safe off the sink loop's own thread. + pending = self._snapshot() + if pending: + _futures_wait(pending, timeout=timeout) + + async def aflush_pending(self, timeout: float = _LOG_FLUSH_TIMEOUT_S) -> None: + pending = self._snapshot() + if not pending: + return + await asyncio.wait([asyncio.wrap_future(f) for f in pending], timeout=timeout) + + +def install_runtime_sinks( + job_id: str, callback: Any, loop: asyncio.AbstractEventLoop +) -> "_IpcLogHandler | 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. Raises if this runtime has no sinks to install into. + """ + try: + from uipath.runtime.output_sinks import ( + set_log_handler, + set_result_sink, + ) + except ImportError as e: + # The caller has already stopped watching execution.log on our promise, so falling back to + # it would lose the whole job's output in silence. + raise RuntimeError( + "This uipath-runtime has no output sinks, so job logs cannot be streamed over IPC. " + "Install uipath-runtime>=0.13.5." + ) from e + + 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 + ) + if future.result(timeout=_SET_RESULT_TIMEOUT_S) is False: + logger.error( + "The handler rejected the job result (SetResult returned false)" + ) + except Exception: + logger.exception("Failed to deliver job result over IPC (SetResult)") + + set_log_handler(handler) + set_result_sink(_result_sink) + return handler + + +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 _stop_loop_thread( + loop: asyncio.AbstractEventLoop, thread: threading.Thread, timeout: float +) -> None: + # Never raise here: a wedged thread leaves the loop running, and close() would mask the job's outcome. + try: + loop.call_soon_threadsafe(loop.stop) + except Exception: + pass + thread.join(timeout=timeout) + if thread.is_alive(): + return + try: + loop.close() + except Exception: + pass + + +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, + log_handler: "_IpcLogHandler | None" = None, + ) -> None: + self._client = client + self._loop = loop + self._thread = thread + self._log_handler = log_handler + + def _shutdown(self) -> None: + """Close the client and stop its loop/thread.""" + if self._log_handler is not None: + self._log_handler.flush_pending() + try: + asyncio.run_coroutine_threadsafe(self._client.aclose(), self._loop).result( + timeout=_SET_RESULT_TIMEOUT_S + ) + except Exception: + pass + _stop_loop_thread(self._loop, self._thread, _SET_RESULT_TIMEOUT_S) + + +def is_wire_job_id(job_id: str | None) -> TypeGuard[str]: + """The peer types the job id as a Guid, and routes nothing for one it can't match.""" + try: + parsed = uuid.UUID(str(job_id)) + except ValueError: + return False + # All-zeros is Guid's default, so it is the one unusable value a caller reaches by omission. + return parsed.int != 0 + + +def connect_handler_ipc(pipe: str, job_id: str | None) -> _HandlerIpcConnection: + """Dial ``pipe`` on a dedicated loop/thread and install the sinks (see ``_HandlerIpcConnection``).""" + if not is_wire_job_id(job_id): + raise RuntimeError( + f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job id; got {job_id!r}." + ) + + 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: + transport = NamedPipeClientTransport(pipe) + # Dial now: execution.log is already suppressed, so a dead pipe would lose every line in silence. + try: + _, writer = await transport.connect() + except Exception as e: + raise RuntimeError( + f"Could not reach the handler IPC pipe {pipe!r}: {e}" + ) from e + writer.close() + client = IpcClient( + transport=transport, + request_timeout=_IPC_REQUEST_TIMEOUT_S, + max_message_size=_MAX_MESSAGE_BYTES, + ) + 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: + _stop_loop_thread(loop, thread, _SET_RESULT_TIMEOUT_S) + raise + # Install in the caller's context: the sinks are contextvars, read in the job's own context at teardown. + handler = install_runtime_sinks(job_id, proxy, loop) + return _HandlerIpcConnection(client, loop, thread, handler) + + +async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None: + """Clear the sinks and tear down the connection.""" + clear_runtime_sinks() + await asyncio.to_thread(conn._shutdown) + + +@contextlib.asynccontextmanager +async def handler_ipc_connection( + pipe: str | None, job_id: str | None +) -> 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 is not None 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..1b70b6cb5 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. That orders them against + the next job's start, but claims nothing once this task is cancelled: the job runs on a thread + that cancellation cannot reach, so it outlives the lock and the globals move under it. + """ 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 ffd89be9c..b94dd85ba 100644 --- a/packages/uipath/src/uipath/_cli/cli_run.py +++ b/packages/uipath/src/uipath/_cli/cli_run.py @@ -116,6 +116,13 @@ 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, + hidden=True, # set by the job executor, never by a person + 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, @@ -130,6 +137,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 @@ -164,6 +172,13 @@ def run( console.error(result.error_message) return + if not result.should_continue and handler_ipc_pipe is not None: + console.error( + "--handler-ipc-pipe was requested, but a plugin took over the run and cannot " + "stream the logs or the result over it." + ) + return + if result.should_continue: try: @@ -213,8 +228,13 @@ async def execute() -> None: JsonLinesFileExporter(ctx.trace_file) ) - async with ResourceOverwritesContext( - lambda: read_resource_overwrites_from_file(ctx.runtime_dir) + from ._job_api import handler_ipc_connection + + async with ( + handler_ipc_connection(handler_ipc_pipe, ctx.job_id), + 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..cd489bd17 100644 --- a/packages/uipath/src/uipath/_cli/cli_server_ipc.py +++ b/packages/uipath/src/uipath/_cli/cli_server_ipc.py @@ -1,9 +1,20 @@ +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 dep: the annotation stays a string forward-ref, so this placeholder is never subscripted. + try: + from uipath_ipc import Message + except ImportError: # pragma: no cover - no IPC means Register is never dispatched + Message = object + console = ConsoleLogger() @@ -23,6 +34,7 @@ class PythonServerRunRequest: Args: str | list[str] | None = None WorkingDirectory: str | None = None EnvironmentVariables: dict[str, str] = field(default_factory=dict) + StreamOutputOverIpc: bool = False @dataclass @@ -42,12 +54,17 @@ 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.""" @abstractmethod - async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobResult: - """Run a job → PythonServerRunJobResult(ExitCode, Error).""" + async def RunJob( + self, request: PythonServerRunRequest, *, message: "Message[None] | None" = None + ) -> PythonServerRunJobResult: + """Run a job → PythonServerRunJobResult(ExitCode, Error). + + ``message`` is injected by the dispatcher, which reads this contract — not the impl. + """ @abstractmethod async def StopJob(self, request: PythonServerStopJobRequest) -> bool: @@ -57,11 +74,13 @@ 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: console.info("Runtime client registered.") return True - async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobResult: + async def RunJob( + self, request: PythonServerRunRequest, *, message: "Message[None] | None" = None + ) -> PythonServerRunJobResult: command_name = request.Command if not isinstance(command_name, str) or not command_name: return PythonServerRunJobResult( @@ -80,9 +99,57 @@ async def RunJob(self, request: PythonServerRunRequest) -> PythonServerRunJobRes f"Running job {_run_id(request.JobKey, request.ResumeVersion)}: {command_name} {args}" ) + on_run_start: "Any" = None + on_run_end: "Any" = None + installed: "list[Any]" = [] + if request.StreamOutputOverIpc: + # Never stored: a captured callback goes stale on reconnect/restart. + from ._job_api import ( + IJobInvocationCommonApi, + clear_runtime_sinks, + install_runtime_sinks, + is_wire_job_id, + ) + + if not is_wire_job_id(request.JobKey): + return PythonServerRunJobResult( + ExitCode=1, + Error=f"StreamOutputOverIpc needs a 'JobKey' that is a job id; got {request.JobKey!r}", + ) + + if message is None or message.client is None: + return PythonServerRunJobResult( + ExitCode=1, + Error="StreamOutputOverIpc is only available when RunJob is invoked over IPC", + ) + + # get_callback only wraps the connection, so this cannot tell us whether the peer + # actually hosts the contract; a peer that doesn't shows up as a failing send. + callback = message.client.get_callback(IJobInvocationCommonApi) # type: ignore[type-abstract] + + loop = asyncio.get_running_loop() + job_key = request.JobKey + + def _install() -> None: + installed.append(install_runtime_sinks(job_key, callback, loop)) + + on_run_start = _install + 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, ) + + # Await, don't block: these share this loop, and the peer may unregister the job once this returns. + for handler in installed: + if handler is not None: + await handler.aflush_pending() + # IPC contract (PythonServerRunJobResult) carries only ExitCode + Error. return PythonServerRunJobResult( ExitCode=result["ExitCode"], Error=result["Error"] @@ -108,10 +175,18 @@ async def start_ipc_server(pipe_name: str) -> None: ) from e _state.init() + + from uipath._cli import _ensure_runtime_initialized + + _ensure_runtime_initialized() + + from ._job_api import _MAX_MESSAGE_BYTES + server = IpcServer( transport=NamedPipeServerTransport(pipe_name), services={IPythonRuntimeServer: PythonRuntimeService()}, request_timeout=None, # jobs are long-running; no server-side timeout + max_message_size=_MAX_MESSAGE_BYTES, ) console.success(f"IPC server listening on pipe '{pipe_name}'") async with server: 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..0e6918818 --- /dev/null +++ b/packages/uipath/tests/cli/test_job_api.py @@ -0,0 +1,689 @@ +"""The uipath-python job-api glue: result mapping and the runtime-sink installer. + +The runtime's real ``output_sinks`` ContextVars are used, with a fresh pair per test: a dict would +record that a sink was installed but not in which context, and the context is the thing that decides +whether the job ever sees it. +""" + +import asyncio +import io +import logging +import os +import sys +import threading +import time +from concurrent.futures import Future +from contextvars import ContextVar +from typing import Any + +import pytest + +from uipath._cli import _job_api +from uipath.runtime.result import UiPathRuntimeStatus + +JOB_ID = "3f2504e0-4f89-11d3-9a0c-0305e82c3301" +JOB_ID_2 = "9a7b1c22-0e54-4f7d-8c31-6b2f0a5d9e10" + + +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 = 404 + + class _Result: + status = UiPathRuntimeStatus.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 + # Every field, so a swapped Title/Detail (a stack trace shown as the error's title in the + # job's failure record) cannot pass. + assert ( + dto.error.Code, + dto.error.Title, + dto.error.Detail, + dto.error.Category, + dto.error.Status, + ) == ("BOOM", "It broke", "stack", "User", 404) + + +def test_to_result_dto_defaults_to_successful_without_error(): + class _Result: + status = UiPathRuntimeStatus.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_result_dto_maps_suspended(): + """Mapping only. Suspend never reaches SetResult -- the runtime delivers SUCCESSFUL/FAULTED + alone and the peer resolves a suspended job from output.json. Do not read this as the + suspend path being carried over IPC.""" + + class _Result: + status = UiPathRuntimeStatus.SUSPENDED + error = None + + dto = _job_api._to_result_dto("j", _Result(), "p.args") + assert dto.status == _job_api.ExecutorJobStatus.SUSPENDED.value + + +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 _isolated_output_sinks(monkeypatch) -> Any: + """Read the sinks back through the REAL ContextVars. + + A dict would record that something was installed but not *where*: installing from the wrong + thread or task would keep every test green while the job, looking the value up in its own + context, found nothing. Fresh ContextVars per test keep that observable and stop one test's + sinks leaking into the next. + """ + from uipath.runtime import output_sinks + + monkeypatch.setattr( + output_sinks, "_log_handler", ContextVar("test_log_handler", default=None) + ) + monkeypatch.setattr( + output_sinks, "_result_sink", ContextVar("test_result_sink", default=None) + ) + + class _Installed: + def __getitem__(self, key: str) -> Any: + if key == "handler": + return output_sinks.get_log_handler() + if key == "sink": + return output_sinks.get_result_sink() + raise KeyError(key) + + return _Installed() + + +def test_install_wires_log_handler_and_result_sink(monkeypatch): + captured = _isolated_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, tagged with the job id. + handler = captured["handler"] + handler.emit( + logging.LogRecord("n", logging.WARNING, "p", 1, "hi %s", ("there",), None) + ) + await handler.aflush_pending() + 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 = UiPathRuntimeStatus.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_an_unformattable_record_does_not_escape_emit(monkeypatch): + """A mid-flight send failure is covered by the broken-channel test; this is the sync path.""" + captured = _isolated_output_sinks(monkeypatch) + + class _Callback: + async def SendLog(self, jid: str, log: Any) -> None: + return None + + loop = asyncio.new_event_loop() + try: + _job_api.install_runtime_sinks(JOB_ID, _Callback(), loop) + handler = captured["handler"] + handled: list[Any] = [] + monkeypatch.setattr(handler, "handleError", handled.append) + # %d against a str: formatting raises inside emit. + handler.emit(logging.LogRecord("n", logging.INFO, "p", 1, "%d", ("x",), None)) + assert len(handled) == 1 + assert handler._snapshot() == [] + finally: + loop.close() + + +def test_transport_and_self_logs_never_ride_the_ipc_channel(monkeypatch): + captured = _isolated_output_sinks(monkeypatch) + sent: list[Any] = [] + forwarded = threading.Event() + + class _Callback: + async def SendLog(self, jid: str, dto: Any) -> None: + sent.append((jid, dto)) + forwarded.set() + + buf = io.StringIO() + monkeypatch.setattr(sys, "__stderr__", buf) + + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + try: + _job_api.install_runtime_sinks("job-6", _Callback(), loop) + handler = captured["handler"] + for name in ("uipath_ipc.client.connection", _job_api.__name__): + handler.emit( + logging.LogRecord( + name, logging.WARNING, "p", 1, "internal chatter", (), None + ) + ) + handler.emit( + logging.LogRecord("job.logger", logging.INFO, "p", 1, "job line", (), None) + ) + forwarded.wait(timeout=5) + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + loop.close() + + assert [dto.Message for _, dto in sent] == ["job line"] + assert buf.getvalue().count("internal chatter") == 2 + + +def test_rejected_result_is_reported(monkeypatch): + captured = _isolated_output_sinks(monkeypatch) + + class _Callback: + async def SetResult(self, jid: str, dto: Any) -> bool: + return False + + class _Result: + status = UiPathRuntimeStatus.SUCCESSFUL + error = None + + records: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + api_logger = logging.getLogger("uipath._cli._job_api") + handler = _Capture() + api_logger.addHandler(handler) + prior_level = api_logger.level + api_logger.setLevel(logging.ERROR) + + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + try: + _job_api.install_runtime_sinks("job-8", _Callback(), loop) + captured["sink"](_Result(), "out.args") + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + loop.close() + api_logger.removeHandler(handler) + api_logger.setLevel(prior_level) + + assert any("rejected" in r.getMessage().lower() for r in records) + + +def test_teardown_does_not_raise_when_the_loop_thread_is_wedged(): + loop = asyncio.new_event_loop() + started = threading.Event() + release = threading.Event() + + def _run() -> None: + asyncio.set_event_loop(loop) + + async def _block() -> None: + started.set() + release.wait(timeout=10) + + loop.run_until_complete(_block()) + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + assert started.wait(timeout=5) + + _job_api._stop_loop_thread(loop, thread, 0.2) + assert thread.is_alive() + + release.set() + thread.join(timeout=5) + loop.close() + + +def test_pending_log_sends_are_flushed_before_teardown(monkeypatch): + _isolated_output_sinks(monkeypatch) + landed: list[str] = [] + + class _Callback: + async def SendLog(self, jid: str, dto: Any) -> None: + await asyncio.sleep(0.2) + landed.append(dto.Message) + + class _Client: + async def aclose(self) -> None: + return None + + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + handler = _job_api.install_runtime_sinks("job-7", _Callback(), loop) + assert handler is not None + handler.emit(logging.LogRecord("j", logging.INFO, "p", 1, "tail line", (), None)) + + conn = _job_api._HandlerIpcConnection(_Client(), loop, thread, handler) + conn._shutdown() + + assert landed == ["tail line"] + + +def test_result_delivery_failure_is_swallowed_and_logged(monkeypatch): + captured = _isolated_output_sinks(monkeypatch) + + class _Callback: + async def SetResult(self, jid: str, dto: Any) -> bool: + raise RuntimeError("handler said no") + + class _Result: + status = UiPathRuntimeStatus.SUCCESSFUL + error = None + + # Capture on the module logger directly, not via caplog: other tests in the suite mutate root + # handlers / propagate, which would silently drop the record from caplog. + records: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + api_logger = logging.getLogger("uipath._cli._job_api") + handler = _Capture() + api_logger.addHandler(handler) + prior_level = api_logger.level + api_logger.setLevel(logging.ERROR) + + # Ack loop on its own thread so the cross-thread wait is deterministic. + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + try: + _job_api.install_runtime_sinks("job-9", _Callback(), loop) + sink = captured["sink"] + sink(_Result(), "out.args") # must not raise + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + loop.close() + api_logger.removeHandler(handler) + api_logger.setLevel(prior_level) + + assert any( + "Failed to deliver job result over IPC" in r.getMessage() for r in records + ) + + +def test_clear_is_a_noop_without_the_runtime(monkeypatch): + # Teardown still has to work on an older runtime -- only install refuses (see the test below). + monkeypatch.setitem(sys.modules, "uipath.runtime.output_sinks", None) + _job_api.clear_runtime_sinks() + + +def test_connect_installs_sinks_and_disconnect_clears(monkeypatch): + pytest.importorskip("uipath_ipc") + captured = _isolated_output_sinks(monkeypatch) + + class _Api(_job_api.IJobInvocationCommonApi): + async def SendLog(self, jobId: str, log: Any) -> None: + return None + + async def SetResult(self, jobId: str, result: Any) -> bool: + return True + + pipe = _unique_jobapi_pipe() + stop = _serve_jobapi_in_background(pipe, _Api()) + try: + + async def scenario() -> None: + # The context manager, not the two halves: cli_run.py only ever uses this, and a + # wiring mistake inside it would leave every assertion below untouched. + async with _job_api.handler_ipc_connection(pipe, JOB_ID): + assert captured["handler"] is not None + assert captured["sink"] is not None + + assert captured["handler"] is None + assert captured["sink"] is None + + asyncio.run(scenario()) + finally: + stop() + + +def test_connect_fails_loudly_when_the_pipe_is_unreachable(monkeypatch): + pytest.importorskip("uipath_ipc") + _isolated_output_sinks(monkeypatch) + + def live() -> int: + return len( + [ + t + for t in threading.enumerate() + if t.name == "uipath-handler-ipc" and t.is_alive() + ] + ) + + before = live() + + with pytest.raises(RuntimeError, match="Could not reach the handler IPC pipe"): + _job_api.connect_handler_ipc("uipath-jobapi-does-not-exist-12345", JOB_ID) + + assert live() == before + + +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_ID) + + +def test_a_broken_log_channel_is_reported_once_to_stderr(monkeypatch): + _isolated_output_sinks(monkeypatch) + + buf = io.StringIO() + monkeypatch.setattr(sys, "__stderr__", buf) + + class _Callback: + async def SendLog(self, job_id: str, log: Any) -> None: + raise RuntimeError("pipe is gone") + + loop = asyncio.new_event_loop() + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + try: + handler = _job_api.install_runtime_sinks(JOB_ID, _Callback(), loop) + assert handler is not None + for i in range(4): + handler.emit( + logging.LogRecord("job", logging.INFO, __file__, 1, f"m{i}", None, None) + ) + handler.flush_pending(timeout=5.0) + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5.0) + loop.close() + + out = buf.getvalue() + assert "no longer reaching the handler" in out, out + assert "pipe is gone" in out + assert out.count("no longer reaching the handler") == 1, out + + +def test_a_cancelled_send_is_reported_like_any_other_failure(monkeypatch): + """`future.exception()` RAISES for a cancelled future, so this arm needs its own case.""" + _isolated_output_sinks(monkeypatch) + buf = io.StringIO() + monkeypatch.setattr(sys, "__stderr__", buf) + + loop = asyncio.new_event_loop() + try: + handler = _job_api._IpcLogHandler(JOB_ID, object(), loop) + cancelled: Future[object] = Future() + assert cancelled.cancel() + + handler._settled(cancelled) + finally: + loop.close() + + assert "no longer reaching the handler" in buf.getvalue() + + +def test_missing_output_sinks_fails_loudly(monkeypatch): + """Silently returning None would lose the job: the peer has already stopped watching the file.""" + monkeypatch.setitem(sys.modules, "uipath.runtime.output_sinks", None) + + with pytest.raises(RuntimeError, match="uipath-runtime"): + _job_api.install_runtime_sinks(JOB_ID, object(), asyncio.new_event_loop()) + + +def test_the_frame_cap_matches_the_dotnet_peer(): + """Both .NET servers on this contract accept 30 MB; uipath-ipc would otherwise default to 2.""" + from uipath_ipc.wire import MAX_PAYLOAD_BYTES + + assert _job_api._MAX_MESSAGE_BYTES == 30 * 1024 * 1024 + assert _job_api._MAX_MESSAGE_BYTES > MAX_PAYLOAD_BYTES + + +def _live_ipc_threads() -> int: + return len( + [ + t + for t in threading.enumerate() + if t.name == "uipath-handler-ipc" and t.is_alive() + ] + ) + + +@pytest.mark.parametrize( + "job_id", + [ + None, + "", + " ", + "job-1", + "not-a-guid", + "00000000-0000-0000-0000-000000000000", # Guid.Empty: parses, routes nowhere + "00000000-0000-0000-0000-00000000000", # a digit short: malformed, not empty + ], +) +def test_connect_without_a_real_job_id_fails_fast(job_id): + before = _live_ipc_threads() + with pytest.raises(RuntimeError, match="UIPATH_JOB_KEY"): + _job_api.connect_handler_ipc("pipe", job_id) + assert _live_ipc_threads() == before + + +@pytest.mark.parametrize("job_id", [None, "", "job-1"]) +async def test_handler_ipc_connection_without_a_real_job_id_fails_fast(job_id): + with pytest.raises(RuntimeError, match="UIPATH_JOB_KEY"): + async with _job_api.handler_ipc_connection("pipe", job_id): + pass + + +_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 = _isolated_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 = UiPathRuntimeStatus.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_ID_2) + # A log line has to survive the round trip too, not just the result. + captured["handler"].emit( + logging.LogRecord( + "job", logging.WARNING, "p", 1, "over %s", ("ipc",), None + ) + ) + # 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. + started = time.monotonic() + captured["sink"](_Result(), "out.args") + elapsed = time.monotonic() - started + + # Assert BEFORE any await. Deadlocked, the sink burns _SET_RESULT_TIMEOUT_S and gives + # up; the queued ack then completes during the await below and the result shows up + # anyway, so anything asserted after an await passes either way. + assert "result" in received, ( + "SetResult did not complete on the caller's loop thread" + ) + assert elapsed < 5, ( + f"the sink blocked for {elapsed:.1f}s — the ack could not run" + ) + + await _job_api.disconnect_handler_ipc(conn) + + asyncio.run(scenario()) + finally: + stop() + + # The log half of the contract, asserted on the wire rather than against a fake. + assert "logs" in received, "SendLog never arrived over the pipe" + log_job_id, entry = received["logs"][0] + assert log_job_id == JOB_ID_2 + assert entry.Message == "over ipc" + assert entry.LogLevel == _job_api.LogLevel.WARNING.value + + assert "result" in received, ( + "SetResult never arrived — the result sink deadlocked/timed out" + ) + job_id, dto = received["result"] + assert job_id == JOB_ID_2 + assert dto.outputArgumentsFilePath == "out.args" + assert dto.status == _job_api.ExecutorJobStatus.SUCCESSFUL.value diff --git a/packages/uipath/tests/cli/test_run.py b/packages/uipath/tests/cli/test_run.py index aa182c7c5..a09b4ac33 100644 --- a/packages/uipath/tests/cli/test_run.py +++ b/packages/uipath/tests/cli/test_run.py @@ -1,5 +1,6 @@ # type: ignore import json +import logging import os from contextlib import asynccontextmanager from unittest.mock import AsyncMock, Mock, patch @@ -178,6 +179,138 @@ def test_run_input_file_success( assert result.exit_code == 0 assert "Successful execution." in result.output + class TestHandlerIpcPipe: + """``--handler-ipc-pipe`` promises the caller logs + result over a pipe.""" + + @staticmethod + def _taken_over(): + return patch( + "uipath._cli.cli_run.Middlewares.next", + return_value=MiddlewareResult( + should_continue=False, + info_message="Execution succeeded", + error_message=None, + should_include_stacktrace=False, + ), + ) + + @pytest.mark.parametrize("pipe", ["some-pipe", ""]) + def test_taken_over_run_cannot_honour_the_pipe( + self, runner: CliRunner, temp_dir: str, entrypoint: str, pipe: str + ): + with runner.isolated_filesystem(temp_dir=temp_dir), self._taken_over(): + result = runner.invoke( + cli, ["run", entrypoint, "--handler-ipc-pipe", pipe] + ) + assert result.exit_code == 1 + assert "a plugin took over the run" in result.output + assert "Successful execution." not in result.output + + def test_taken_over_run_without_the_pipe_is_untouched( + self, runner: CliRunner, temp_dir: str, entrypoint: str + ): + with runner.isolated_filesystem(temp_dir=temp_dir), self._taken_over(): + result = runner.invoke(cli, ["run", entrypoint]) + assert result.exit_code == 0 + assert "Successful execution." in result.output + + def test_the_flag_is_hidden_but_still_accepted(self, runner: CliRunner): + """Handler-only: it must stay out of --help and the published docs, yet keep working.""" + assert ( + "handler-ipc-pipe" not in runner.invoke(cli, ["run", "--help"]).output + ) + # Hiding an option must not un-declare it. + assert ( + "No such option" + not in runner.invoke( + cli, ["run", "x", "--handler-ipc-pipe", "some-pipe"] + ).output + ) + + def test_the_job_runs_inside_the_open_connection( + self, runner: CliRunner, temp_dir: str, monkeypatch + ): + """Dialling the pipe is not enough: the run has to happen while it is open.""" + job_key = "3f2504e0-4f89-11d3-9a0c-0305e82c3301" + monkeypatch.setenv("UIPATH_JOB_KEY", job_key) + + events: list[str] = [] + seen: list[tuple[str, str]] = [] + snapshots: list[object] = [] + + from uipath.runtime import context as runtime_context + from uipath.runtime import output_sinks + + sentinel = logging.Handler() + + @asynccontextmanager + async def _fake_connection(pipe, job_id): + seen.append((pipe, job_id)) + events.append("open") + # Install for real: the ordering that matters is against the runtime context's + # one-shot snapshot, not against runtime.execute. + output_sinks.set_log_handler(sentinel) + try: + yield None + finally: + output_sinks.set_log_handler(None) + events.append("close") + + # UiPathRuntimeContext.__enter__ reads the sink exactly once; record what it saw. + real_get = runtime_context.get_log_handler + + def _spy_get_log_handler(): + value = real_get() + snapshots.append(value) + return value + + monkeypatch.setattr( + runtime_context, "get_log_handler", _spy_get_log_handler + ) + + factory = _make_mock_factory(["my_agent"]) + runtime = factory.new_runtime.return_value + + async def _execute(*a, **k): + events.append("run") + return Mock(status="SUCCESSFUL") + + runtime.execute = AsyncMock(side_effect=_execute) + + with runner.isolated_filesystem(temp_dir=temp_dir): + with ( + patch( + "uipath._cli.cli_run.Middlewares.next", + return_value=_middleware_continue(), + ), + patch( + "uipath._cli.cli_run.UiPathRuntimeFactoryRegistry.get", + return_value=factory, + ), + patch( + "uipath._cli.cli_run.ResourceOverwritesContext", + side_effect=_mock_resource_overwrites_context, + ), + patch( + "uipath._cli._job_api.handler_ipc_connection", _fake_connection + ), + ): + result = runner.invoke( + cli, ["run", "--handler-ipc-pipe", "some-pipe"] + ) + + assert result.exit_code == 0, f"output: {result.output!r}" + # The whole point: the connection must still be open when the job runs. + assert events == ["open", "run", "close"], events + # And it must be told which job, not None. + assert seen == [("some-pipe", job_key)] + # The connection must be open BEFORE the runtime context snapshots the sinks: move it + # inside and __enter__ would capture None, losing every log line and the result. + assert snapshots == [sentinel], ( + "the runtime context did not see the installed log sink" + ) + assert "took over the run" not in result.output + class TestMiddleware: def test_autodiscover_entrypoint(self, runner: CliRunner, temp_dir: str): """When exactly one entrypoint exists, it is auto-resolved.""" diff --git a/packages/uipath/tests/cli/test_server_ipc.py b/packages/uipath/tests/cli/test_server_ipc.py index 801920dd2..711e0fd58 100644 --- a/packages/uipath/tests/cli/test_server_ipc.py +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -13,24 +13,28 @@ import asyncio import json +import logging import os import sys import threading import time -from typing import Any, Awaitable, Callable +from typing import Any, Awaitable, Callable, cast import click import pytest from uipath_ipc import ( IpcClient, IpcServer, + Message, NamedPipeClientTransport, NamedPipeServerTransport, ) from uipath._cli import _server_core +from uipath._cli._job_api import IJobInvocationCommonApi from uipath._cli.cli_server import ( IPythonRuntimeServer, + PythonRuntimeService, PythonServerRunJobResult, start_ipc_server, ) @@ -124,6 +128,9 @@ def test_start_ipc_server_fails_fast_without_uipath_ipc(monkeypatch): asyncio.run(coro) +JOB_ID = "3f2504e0-4f89-11d3-9a0c-0305e82c3301" + + class TestIpcServer: @pytest.fixture def pipe(self): @@ -284,10 +291,12 @@ 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: + async def RunJob( + self, request: Any, *, message: Any = None + ) -> PythonServerRunJobResult: received.append(request) return PythonServerRunJobResult(ExitCode=0) @@ -329,3 +338,309 @@ 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 is a readiness probe; RunJob derives the callback per request and installs sinks.""" + + def test_register_is_a_readiness_probe(self): + service = PythonRuntimeService() + + async def scenario() -> None: + assert await service.Register(Message(client=cast(Any, object()))) is True + + asyncio.run(scenario()) + + 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()) + + def test_runjob_fails_loud_when_not_invoked_over_ipc(self, monkeypatch): + """Only reachable by a direct call: the dispatcher always injects a Message with a client.""" + from uipath._cli import cli_server_ipc + from uipath._cli.cli_server_ipc import PythonServerRunRequest + + ran: list[bool] = [] + + async def _fake_run(*args: Any, **kwargs: Any) -> dict[str, Any]: + ran.append(True) + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) + + message: "Message[None]" + message = Message() + + service = PythonRuntimeService() + request = PythonServerRunRequest( + JobKey=JOB_ID, Command="run", Args=[], StreamOutputOverIpc=True + ) + + async def scenario() -> Any: + return await service.RunJob(request, message=message) + + result = asyncio.run(scenario()) + assert result.ExitCode == 1 + assert "over IPC" in (result.Error or "") + assert ran == [] + + @pytest.mark.parametrize( + "job_key", ["", "job-9", "00000000-0000-0000-0000-000000000000"] + ) + def test_runjob_without_a_real_job_key_fails_loudly(self, monkeypatch, job_key): + from uipath._cli import _job_api, cli_server_ipc + from uipath._cli.cli_server_ipc import ( + PythonRuntimeService, + PythonServerRunRequest, + ) + + events: list[Any] = [] + monkeypatch.setattr( + _job_api, + "install_runtime_sinks", + lambda jid, cb, loop: events.append(("install", jid)), + ) + + async def _fake_run(*args: Any, **kwargs: Any) -> dict[str, Any]: + events.append("run") + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) + + class _Client: + def get_callback(self, contract: Any) -> Any: + return "CALLBACK" + + service = PythonRuntimeService() + request = PythonServerRunRequest( + JobKey=job_key, Command="run", Args=[], StreamOutputOverIpc=True + ) + + async def scenario() -> Any: + return await service.RunJob( + request, message=Message(client=cast(Any, _Client())) + ) + + result = asyncio.run(scenario()) + assert result.ExitCode == 1 + assert "JobKey" in (result.Error or "") + assert events == [] + + def test_runjob_without_streaming_does_not_need_a_job_key(self, monkeypatch): + from uipath._cli import cli_server_ipc + from uipath._cli.cli_server_ipc import ( + PythonRuntimeService, + PythonServerRunRequest, + ) + + ran: list[bool] = [] + + async def _fake_run(*args: Any, **kwargs: Any) -> dict[str, Any]: + ran.append(True) + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) + + service = PythonRuntimeService() + request = PythonServerRunRequest(JobKey="", Command="run", Args=[]) + + async def scenario() -> Any: + return await service.RunJob(request) + + result = asyncio.run(scenario()) + assert result.ExitCode == 0 + assert ran == [True] + + def test_pooled_streaming_works_over_a_real_pipe(self, monkeypatch): + """No hand-fed ``message``: the dispatcher must inject it from the contract.""" + from uipath._cli import _job_api, cli_server_ipc + + installed: list[Any] = [] + real_install = _job_api.install_runtime_sinks + + def _record(jid, cb, loop): + installed.append(jid) + # Install for real: patching this away would leave the callback unused, so nothing + # would notice if RunJob asked the peer for the wrong contract. + return real_install(jid, cb, loop) + + monkeypatch.setattr(_job_api, "install_runtime_sinks", _record) + + async def _fake_run(cmd, args, env, wd, on_run_start=None, on_run_end=None): + if on_run_start: + on_run_start() + # Emit through the installed sink, so a log line actually crosses the pipe. + from uipath.runtime import output_sinks + + handler = cast(Any, output_sinks.get_log_handler()) + assert handler is not None, "the sinks were not installed for this job" + handler.emit( + logging.LogRecord( + "job", logging.INFO, "p", 1, "pooled %s", ("line",), None + ) + ) + handler.flush_pending(timeout=10.0) + if on_run_end: + on_run_end() + return {"ExitCode": 0, "Error": None} + + monkeypatch.setattr(cli_server_ipc, "_run_command_isolated", _fake_run) + + delivered: list[tuple[str, Any]] = [] + + class _Callback: + async def SendLog(self, job_id: str, log: Any) -> None: + delivered.append((job_id, log)) + + async def SetResult(self, job_id: str, result: Any) -> bool: + return True + + pipe = _unique_pipe() + _serve_in_background(pipe) + + async def scenario() -> Any: + client = IpcClient( + transport=NamedPipeClientTransport(pipe), + callbacks={IJobInvocationCommonApi: _Callback()}, + ) + try: + proxy = client.get_proxy(IPythonRuntimeServer) # type: ignore[type-abstract] + return await proxy.RunJob( + cast( + Any, + { + "JobKey": JOB_ID, + "Command": "run", + "Args": [], + "StreamOutputOverIpc": True, + }, + ) + ) + finally: + await client.aclose() + + result = asyncio.run(scenario()) + assert result.Error is None, result.Error + assert result.ExitCode == 0 + assert installed == [JOB_ID] + # The contract passed to get_callback IS the endpoint key on the wire: ask for the wrong + # one and every send is addressed to something the peer does not host. + assert len(delivered) == 1, "no log line crossed the pooled callback" + assert delivered[0][0] == JOB_ID + assert delivered[0][1].Message == "pooled line" + + def test_runjob_installs_the_sinks_from_the_request_callback(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): + 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) + + class _Client: + def get_callback(self, contract: Any) -> Any: + return "CALLBACK" + + service = PythonRuntimeService() + request = PythonServerRunRequest( + JobKey=JOB_ID, Command="run", Args=[], StreamOutputOverIpc=True + ) + + async def scenario() -> None: + await service.RunJob(request, message=Message(client=_Client())) + + asyncio.run(scenario()) + assert events == [("install", JOB_ID, "CALLBACK"), ("run",), ("clear",)] + + def test_runjob_drains_pending_sends_before_returning(self, monkeypatch): + from uipath._cli import _job_api, cli_server_ipc + from uipath._cli.cli_server_ipc import PythonServerRunRequest + + events: list[str] = [] + + class _Handler: + async def aflush_pending(self, *a: Any, **k: Any) -> None: + events.append("flush") + + monkeypatch.setattr( + _job_api, "install_runtime_sinks", lambda jid, cb, loop: _Handler() + ) + monkeypatch.setattr(_job_api, "clear_runtime_sinks", lambda: None) + + async def _fake_run(cmd, args, env, wd, on_run_start=None, on_run_end=None): + 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) + + class _Client: + def get_callback(self, contract: Any) -> Any: + return "CALLBACK" + + service = PythonRuntimeService() + request = PythonServerRunRequest( + JobKey=JOB_ID, Command="run", Args=[], StreamOutputOverIpc=True + ) + + async def scenario() -> None: + await service.RunJob(request, message=Message(client=_Client())) + + asyncio.run(scenario()) + assert events == ["run", "flush"] + + 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): + 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() + request = PythonServerRunRequest(JobKey="job-9", Command="run", Args=[]) + + async def scenario() -> None: + await service.RunJob(request, message=Message(client=None)) + + asyncio.run(scenario()) + assert events == [("run",)] diff --git a/packages/uipath/tests/cli/test_server_job_core.py b/packages/uipath/tests/cli/test_server_job_core.py index 1722d0c5b..955099d2a 100644 --- a/packages/uipath/tests/cli/test_server_job_core.py +++ b/packages/uipath/tests/cli/test_server_job_core.py @@ -68,6 +68,75 @@ async def test_reports_unexpected_exception(restore_state: Any) -> None: assert "boom" in result["Error"] +# The pooled channel installs a per-job sink via on_run_start and clears it via +# on_run_end. The IPC layer's own tests fake this core, so the placement guarantees +# below (clear-on-raise, install-after-cwd, no hooks on the bad-cwd early return) +# are pinned here against the real core. + + +@pytest.mark.parametrize("boom", [SystemExit(2), ValueError("kaboom")]) +async def test_on_run_end_fires_even_when_the_job_raises( + restore_state: Any, boom: BaseException +) -> None: + _init(restore_state) + events: list[str] = [] + cmd = Mock() + cmd.main.side_effect = boom + await _server_core._run_command_isolated( + cmd, + [], + {}, + None, + on_run_start=lambda: events.append("start"), + on_run_end=lambda: events.append("end"), + ) + # on_run_end is in a finally, so a raising job still clears its sink before the + # next job runs — otherwise a stale sink would cross-wire the next job. + assert events == ["start", "end"] + + +async def test_on_run_start_sees_the_jobs_cwd_and_env( + restore_state: Any, tmp_path: Any +) -> None: + _init(restore_state) + seen: dict[str, Any] = {} + cmd = Mock() + + def _start() -> None: + seen["cwd"] = os.getcwd() + seen["env"] = os.environ.get("JOB_VAR") + + await _server_core._run_command_isolated( + cmd, + [], + {"JOB_VAR": "42"}, + str(tmp_path), + on_run_start=_start, + on_run_end=lambda: None, + ) + # The hook runs after env/cwd are applied, so it sees exactly what the job sees. + assert os.path.samefile(seen["cwd"], str(tmp_path)) + assert seen["env"] == "42" + + +async def test_hooks_do_not_run_on_bad_working_dir( + restore_state: Any, tmp_path: Any +) -> None: + _init(restore_state) + events: list[str] = [] + result = await _server_core._run_command_isolated( + Mock(), + [], + {}, + str(tmp_path / "does-not-exist"), + on_run_start=lambda: events.append("start"), + on_run_end=lambda: events.append("end"), + ) + assert result["ClientError"] is True + # The early return sits above on_run_start: no install means no clear owed. + assert events == [] + + # parse_args accepts what every caller sends: the .NET peer sends a single # string (shlex-split), HTTP dicts / tests may send a pre-split list, or None. diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index 98877ec72..4483eaf38 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2599,7 +2599,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.14.16" +version = "2.14.17" source = { editable = "." } dependencies = [ { name = "applicationinsights" }, @@ -2683,7 +2683,7 @@ requires-dist = [ { name = "uipath-core", editable = "../uipath-core" }, { name = "uipath-ipc", marker = "extra == 'ipc'", specifier = ">=2.5.1,<2.6.0" }, { name = "uipath-platform", editable = "../uipath-platform" }, - { name = "uipath-runtime", specifier = ">=0.13.1,<0.14.0" }, + { name = "uipath-runtime", specifier = ">=0.13.5,<0.14.0" }, ] provides-extras = ["ipc"] @@ -2802,16 +2802,16 @@ dev = [ [[package]] name = "uipath-runtime" -version = "0.13.1" +version = "0.13.5" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "chardet" }, { name = "uipath-core" }, { name = "vadersentiment" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/15/42/837240fd4cb2824ee5899cd2e06352c617645077efcecff7081a7292464a/uipath_runtime-0.13.1.tar.gz", hash = "sha256:4fd51d95e8b3c8554de8148239425499bbb7ec8687ad93a38f5107c6cdc0d7dd", size = 246257, upload-time = "2026-08-12T12:48:20.386Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c8/8f/14ff9ccf152b6d97c0881f6b40d3e01037045ee52c467ed00dbb2319a0cb/uipath_runtime-0.13.5.tar.gz", hash = "sha256:923e4514f3d868f3711211fee6e2dd39f76c6aaf7354fec9e80cdc5e6fad5d80", size = 250415, upload-time = "2026-09-10T09:32:53.126Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/84/de/94e1ebcd7f25ee7620cb712f376cd84349d5d4b599bdbc6298556310598f/uipath_runtime-0.13.1-py3-none-any.whl", hash = "sha256:365134dfdc87050636b9027b839bbd93636ee56bd463bb38c1e48b7e19e848dd", size = 96876, upload-time = "2026-08-12T12:48:19.014Z" }, + { url = "https://files.pythonhosted.org/packages/95/88/b851e75b3c70bbedde1e24d5b5f38035863ed3d50e8facb75ac7d2db40b8/uipath_runtime-0.13.5-py3-none-any.whl", hash = "sha256:d5eb7d578689f67d5dd631d28a092954c9a8797e130e16fe46e2895e4b0656ac", size = 98513, upload-time = "2026-09-10T09:32:51.404Z" }, ] [[package]]