From 7e1e607137b18b7f0a4a3b30d16a14c98962c121 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Mon, 7 Sep 2026 11:14:51 +0200 Subject: [PATCH 1/2] feat: stream job logs to the handler over uipath-ipc Add a jobapi package that forwards a job's execution logs to the Robot handler over uipath-ipc instead of execution.log, for both non-pooled and pooled executors. Logs only; the result stays on output.json (kept off-heap by split_output_arguments). - contract.py: IIpcLogSink (SendLog) + JobLogDto{Message, LogLevel}, mirroring .NET. - client.py: IpcJobApiClient, the non-pooled per-job client (private asyncio thread, Proactor on Windows, FIFO queue with retry/mute). - pooled.py + log_handler.py: a process-global PooledLogSink and PooledIpcSendLogHandler for the pooled server, which owns the CoreIPC callback one layer up. - context.py: use the IPC handler when UIPATH_JOB_API_IPC_ENDPOINT + UIPATH_JOB_ID are set (non-pooled), or the pooled sink when only UIPATH_JOB_ID is set and a sink is registered; execution.log is suppressed in both cases. Optional [ipc] extra (uipath-ipc). Bump 0.13.4 -> 0.13.5. Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 8 +- src/uipath/runtime/context.py | 52 +++++++ src/uipath/runtime/jobapi/__init__.py | 29 ++++ src/uipath/runtime/jobapi/client.py | 184 +++++++++++++++++++++++ src/uipath/runtime/jobapi/contract.py | 39 +++++ src/uipath/runtime/jobapi/log_handler.py | 61 ++++++++ src/uipath/runtime/jobapi/pooled.py | 28 ++++ tests/test_context_ipc.py | 163 ++++++++++++++++++++ tests/test_jobapi_client.py | 152 +++++++++++++++++++ tests/test_jobapi_contract.py | 41 +++++ tests/test_jobapi_log_handler.py | 57 +++++++ tests/test_jobapi_pooled.py | 66 ++++++++ uv.lock | 21 ++- 13 files changed, 899 insertions(+), 2 deletions(-) create mode 100644 src/uipath/runtime/jobapi/__init__.py create mode 100644 src/uipath/runtime/jobapi/client.py create mode 100644 src/uipath/runtime/jobapi/contract.py create mode 100644 src/uipath/runtime/jobapi/log_handler.py create mode 100644 src/uipath/runtime/jobapi/pooled.py create mode 100644 tests/test_context_ipc.py create mode 100644 tests/test_jobapi_client.py create mode 100644 tests/test_jobapi_contract.py create mode 100644 tests/test_jobapi_log_handler.py create mode 100644 tests/test_jobapi_pooled.py diff --git a/pyproject.toml b/pyproject.toml index c624e4c..95c926d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath-runtime" -version = "0.13.4" +version = "0.13.5" description = "Runtime abstractions and interfaces for building agents and automation scripts in the UiPath ecosystem" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" @@ -20,6 +20,10 @@ maintainers = [ { name = "Cristian Pufu", email = "cristian.pufu@uipath.com" }, ] +[project.optional-dependencies] +# Job-API IPC log channel; required when UIPATH_JOB_API_IPC_ENDPOINT is set. +ipc = ["uipath-ipc>=2.5.1, <2.6.0"] + [project.urls] Homepage = "https://uipath.com" Repository = "https://github.com/UiPath/uipath-runtime-python" @@ -42,6 +46,7 @@ dev = [ "pytest-cov>=4.1.0", "pytest-mock>=3.11.1", "pre-commit>=4.1.0", + "uipath-ipc>=2.5.1, <2.6.0", ] [tool.hatch.build.targets.wheel] @@ -123,6 +128,7 @@ exclude-newer = "2 days" [tool.uv.exclude-newer-package] uipath-core = false +uipath-ipc = false [[tool.uv.index]] name = "testpypi" diff --git a/src/uipath/runtime/context.py b/src/uipath/runtime/context.py index 575db03..91f8fb9 100644 --- a/src/uipath/runtime/context.py +++ b/src/uipath/runtime/context.py @@ -18,6 +18,9 @@ UiPathErrorContract, UiPathRuntimeError, ) +from uipath.runtime.jobapi.client import IpcJobApiClient +from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, PooledIpcSendLogHandler +from uipath.runtime.jobapi.pooled import get_pooled_log_sink from uipath.runtime.logging._interceptor import UiPathRuntimeLogsInterceptor from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus @@ -120,9 +123,38 @@ class UiPathRuntimeContext(BaseModel): keep_state_file: bool = Field( False, description="Prevents deletion of state file before running." ) + ipc_endpoint: str | None = Field( + None, + description=( + "uipath-ipc endpoint (UIPATH_JOB_API_IPC_ENDPOINT) for streaming logs to " + "the handler in place of execution.log. The result stays on output.json." + ), + ) + ipc_job_id: str | None = Field( + None, + description="Handler job id (UIPATH_JOB_ID) used to route the IPC calls.", + ) + ipc_client: Any = Field(default=None, exclude=True, repr=False) model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") + @property + def ipc_active(self) -> bool: + """Whether logs flow over a per-job IPC pipe (non-pooled) instead of to files.""" + return bool(self.ipc_endpoint and self.ipc_job_id) + + @property + def pooled_ipc_active(self) -> bool: + """Whether logs stream over the pooled server's callback rather than a per-job pipe. + + True when there is a job id but no endpoint, and the pooled server registered its sink. + """ + return ( + bool(self.ipc_job_id) + and not self.ipc_endpoint + and get_pooled_log_sink() is not None + ) + def _apply_execution_source(self) -> None: """Derive execution_source from the command, if not already set. @@ -240,11 +272,27 @@ def __enter__(self): """ # Intercept all stdout/stderr/logs # Write to file (runtime), stdout (debug) or log handler (if provided) + log_handler: logging.Handler | None = None + if self.ipc_active: + assert self.ipc_endpoint is not None and self.ipc_job_id is not None + self.ipc_client = IpcJobApiClient( + self.ipc_endpoint, self.ipc_job_id, logger + ) + self.ipc_client.start() + log_handler = IpcSendLogHandler(self.ipc_client) + log_handler.setFormatter(logging.Formatter("%(message)s")) + elif self.pooled_ipc_active: + sink = get_pooled_log_sink() + assert self.ipc_job_id is not None and sink is not None + log_handler = PooledIpcSendLogHandler(self.ipc_job_id, sink) + log_handler.setFormatter(logging.Formatter("%(message)s")) + self.logs_interceptor = UiPathRuntimeLogsInterceptor( min_level=self.logs_min_level, dir=self.runtime_dir, file=self.logs_file, job_id=self.job_id, + log_handler=log_handler, ) self.logs_interceptor.setup() @@ -354,6 +402,8 @@ def __exit__(self, exc_type, exc_val, exc_tb): # Restore original logging if hasattr(self, "logs_interceptor"): self.logs_interceptor.teardown() + if self.ipc_client is not None: + self.ipc_client.close() @cached_property def resolved_result_file_path(self) -> str: @@ -418,6 +468,8 @@ def with_defaults( base.tenant_id = os.environ.get("UIPATH_TENANT_ID") base.process_key = os.environ.get("UIPATH_PROCESS_UUID") base.folder_key = os.environ.get("UIPATH_FOLDER_KEY") + base.ipc_endpoint = os.environ.get("UIPATH_JOB_API_IPC_ENDPOINT") + base.ipc_job_id = os.environ.get("UIPATH_JOB_ID") # Override with kwargs for k, v in kwargs.items(): diff --git a/src/uipath/runtime/jobapi/__init__.py b/src/uipath/runtime/jobapi/__init__.py new file mode 100644 index 0000000..8945ac2 --- /dev/null +++ b/src/uipath/runtime/jobapi/__init__.py @@ -0,0 +1,29 @@ +"""Job-API IPC: stream job logs back to the handler (per-job client or pooled callback).""" + +from uipath.runtime.jobapi.client import IpcJobApiClient +from uipath.runtime.jobapi.contract import ( + IIpcLogSink, + JobLogDto, + LogLevel, +) +from uipath.runtime.jobapi.log_handler import ( + IpcSendLogHandler, + PooledIpcSendLogHandler, +) +from uipath.runtime.jobapi.pooled import ( + PooledLogSink, + get_pooled_log_sink, + set_pooled_log_sink, +) + +__all__ = [ + "IIpcLogSink", + "IpcJobApiClient", + "IpcSendLogHandler", + "JobLogDto", + "LogLevel", + "PooledIpcSendLogHandler", + "PooledLogSink", + "get_pooled_log_sink", + "set_pooled_log_sink", +] diff --git a/src/uipath/runtime/jobapi/client.py b/src/uipath/runtime/jobapi/client.py new file mode 100644 index 0000000..e117ec7 --- /dev/null +++ b/src/uipath/runtime/jobapi/client.py @@ -0,0 +1,184 @@ +"""uipath-ipc client that streams job logs to the handler (logs only).""" + +from __future__ import annotations + +import asyncio +import importlib.util +import logging +import sys +import threading +import time +from typing import Any + +from uipath.runtime.jobapi.contract import IIpcLogSink, JobLogDto + +_SEND_TIMEOUT_S = 1.0 +_MAX_SEND_ATTEMPTS = 3 +_RETRY_DELAY_S = 0.2 +_MAX_CONSECUTIVE_DROPS = 3 +_FAILURE_COOLDOWN_S = 30.0 +_STARTUP_TIMEOUT_S = 10.0 +_SHUTDOWN_TIMEOUT_S = 15.0 + +_STOP = object() + + +class IpcJobApiClient: + """Owns a uipath-ipc client on a private event-loop thread. + + The runtime's ``__exit__`` is synchronous but IPC is asyncio-based, so the + client runs its loop on its own thread (a Proactor loop on Windows, which + named pipes require). Logs are enqueued without blocking the caller and + drained one at a time, FIFO; the queue is flushed at job end. A send that + fails is retried a few times, then that one entry is dropped, and after + repeated failures forwarding is muted for a cooldown — mirroring the JS + coded-functions log forwarder. + """ + + def __init__(self, endpoint: str, job_id: str, log: logging.Logger) -> None: + """Configure the client; call ``start()`` to connect.""" + self._endpoint = endpoint + self._job_id = job_id + self._log = log + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._queue: asyncio.Queue[Any] | None = None + self._proxy: IIpcLogSink | None = None + self._client: Any = None + self._consumer: asyncio.Task[None] | None = None + self._ready = threading.Event() + self._start_error: BaseException | None = None + self._stopping = False + + def start(self) -> None: + """Start the loop thread and connect; raise if that fails or times out.""" + if importlib.util.find_spec("uipath_ipc") is None: + raise RuntimeError( + "Job-API IPC was requested (UIPATH_JOB_API_IPC_ENDPOINT is set) but " + "the 'uipath-ipc' package is not installed. Install it (e.g. " + "'pip install uipath-ipc') to stream logs over IPC." + ) + + self._thread = threading.Thread( + target=self._run, name="uipath-jobapi-ipc", daemon=True + ) + self._thread.start() + if not self._ready.wait(timeout=_STARTUP_TIMEOUT_S): + raise RuntimeError( + f"Job-API IPC client did not connect within {_STARTUP_TIMEOUT_S}s " + f"(endpoint {self._endpoint!r})." + ) + if self._start_error is not None: + raise self._start_error + + def _run(self) -> None: + try: + if sys.platform == "win32": + self._loop = asyncio.ProactorEventLoop() + else: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + self._loop.run_until_complete(self._setup()) + except BaseException as e: + self._start_error = e + self._ready.set() + return + self._ready.set() + try: + self._loop.run_forever() + finally: + self._loop.close() + + async def _setup(self) -> None: + from uipath_ipc import IpcClient, NamedPipeClientTransport + + self._client = IpcClient( + transport=NamedPipeClientTransport(self._endpoint), + request_timeout=None, + ) + self._proxy = self._client.get_proxy(IIpcLogSink) + self._queue = asyncio.Queue() + self._consumer = asyncio.ensure_future(self._consume()) + + async def _consume(self) -> None: + assert self._queue is not None + consecutive_drops = 0 + muted_until = 0.0 + while True: + item = await self._queue.get() + try: + if item is _STOP: + return + if time.monotonic() < muted_until: + continue + if await self._send_log_once(item): + consecutive_drops = 0 + else: + consecutive_drops += 1 + if consecutive_drops >= _MAX_CONSECUTIVE_DROPS: + muted_until = time.monotonic() + _FAILURE_COOLDOWN_S + consecutive_drops = 0 + self._log.warning( + "Job-API IPC: log forwarding paused for %ss after " + "repeated send failures.", + int(_FAILURE_COOLDOWN_S), + ) + finally: + self._queue.task_done() + + async def _send_log_once(self, dto: JobLogDto) -> bool: + assert self._proxy is not None + last_err: BaseException | None = None + for attempt in range(1, _MAX_SEND_ATTEMPTS + 1): + try: + await asyncio.wait_for( + self._proxy.SendLog(self._job_id, dto), timeout=_SEND_TIMEOUT_S + ) + return True + except Exception as e: + last_err = e + if attempt < _MAX_SEND_ATTEMPTS: + await asyncio.sleep(_RETRY_DELAY_S) + self._log.debug("Job-API IPC: dropped one log entry: %s", last_err) + return False + + def send_log(self, dto: JobLogDto) -> None: + """Enqueue one log entry for sending (thread-safe, never blocks).""" + loop = self._loop + queue = self._queue + if loop is None or queue is None or self._stopping: + return + try: + loop.call_soon_threadsafe(queue.put_nowait, dto) + except RuntimeError: + pass + + def close(self) -> None: + """Flush queued logs, close the client, and join the loop thread.""" + self._stopping = True + loop = self._loop + thread = self._thread + if loop is None or thread is None or not thread.is_alive(): + return + try: + fut = asyncio.run_coroutine_threadsafe(self._shutdown(), loop) + fut.result(timeout=_SHUTDOWN_TIMEOUT_S) + except Exception as e: + self._log.debug("Job-API IPC: error during shutdown: %s", e) + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=_SHUTDOWN_TIMEOUT_S) + + async def _shutdown(self) -> None: + if self._queue is not None and self._consumer is not None: + await self._queue.join() + self._queue.put_nowait(_STOP) + try: + await asyncio.wait_for(self._consumer, timeout=_SHUTDOWN_TIMEOUT_S) + except Exception: + self._consumer.cancel() + if self._client is not None: + try: + await self._client.aclose() + except Exception as e: + self._log.debug("Job-API IPC: error closing client: %s", e) diff --git a/src/uipath/runtime/jobapi/contract.py b/src/uipath/runtime/jobapi/contract.py new file mode 100644 index 0000000..59a44b1 --- /dev/null +++ b/src/uipath/runtime/jobapi/contract.py @@ -0,0 +1,39 @@ +"""Python mirror of the .NET ``IIpcLogSink`` CoreIpc contract (logs only). + +uipath-ipc routes by the contract class ``__name__`` and method name and serializes each +argument by its declared field names, so the class name, the method name, and every field +below must match the .NET side exactly. The result is not sent over IPC — it stays on +``output.json``, which the handler still reads. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from enum import IntEnum + + +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 + + +@dataclass +class JobLogDto: + """A single log entry (PascalCase fields to match the wire).""" + + Message: str = "" + LogLevel: int = LogLevel.INFORMATION.value + + +class IIpcLogSink(ABC): + """The handler's log-sink contract; the class name is the CoreIpc endpoint key.""" + + @abstractmethod + async def SendLog(self, jobId: str, log: JobLogDto) -> None: + """Forward one log entry (one-way).""" diff --git a/src/uipath/runtime/jobapi/log_handler.py b/src/uipath/runtime/jobapi/log_handler.py new file mode 100644 index 0000000..a7642bd --- /dev/null +++ b/src/uipath/runtime/jobapi/log_handler.py @@ -0,0 +1,61 @@ +"""Logging handlers that forward records to the handler over uipath-ipc.""" + +import logging + +from uipath.runtime.jobapi.client import IpcJobApiClient +from uipath.runtime.jobapi.contract import JobLogDto, LogLevel +from uipath.runtime.jobapi.pooled import PooledLogSink + + +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 + + +class IpcSendLogHandler(logging.Handler): + """Forwards each record to the handler's IIpcLogSink via the non-pooled per-job client.""" + + def __init__(self, client: IpcJobApiClient) -> None: + """Wrap the IPC client the records are forwarded through.""" + super().__init__() + self._client = client + + def emit(self, record: logging.LogRecord) -> None: + """Format the record and hand it to the IPC client (non-blocking).""" + try: + message = self.format(record) + self._client.send_log( + JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno)) + ) + except Exception: + self.handleError(record) + + +class PooledIpcSendLogHandler(logging.Handler): + """Forwards each record to the pooled process-global sink, tagged with the job id.""" + + def __init__(self, job_id: str, sink: PooledLogSink) -> None: + """Bind the job id every record is tagged with and the sink to forward through.""" + super().__init__() + self._job_id = job_id + self._sink = sink + + def emit(self, record: logging.LogRecord) -> None: + """Format the record and hand it to the pooled sink (non-blocking).""" + try: + message = self.format(record) + self._sink( + self._job_id, + JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno)), + ) + except Exception: + self.handleError(record) diff --git a/src/uipath/runtime/jobapi/pooled.py b/src/uipath/runtime/jobapi/pooled.py new file mode 100644 index 0000000..9a61740 --- /dev/null +++ b/src/uipath/runtime/jobapi/pooled.py @@ -0,0 +1,28 @@ +"""Process-global log sink for the pooled path. + +In pooled mode the job runs in-process inside ``uipath server``, which owns the CoreIpc +callback to the handler. The runtime can't reach that callback directly, so the pooled +server registers a sink here and the runtime's log handler forwards to it. The sink is +called from the job's logging (a worker thread), so it must be non-blocking and thread-safe. +""" + +from __future__ import annotations + +from typing import Callable + +from uipath.runtime.jobapi.contract import JobLogDto + +PooledLogSink = Callable[[str, JobLogDto], None] + +_sink: "PooledLogSink | None" = None + + +def set_pooled_log_sink(sink: "PooledLogSink | None") -> None: + """Register (or clear, with ``None``) the process-global pooled log sink.""" + global _sink + _sink = sink + + +def get_pooled_log_sink() -> "PooledLogSink | None": + """The registered pooled log sink, or None when not in a pooled server.""" + return _sink diff --git a/tests/test_context_ipc.py b/tests/test_context_ipc.py new file mode 100644 index 0000000..77eacfe --- /dev/null +++ b/tests/test_context_ipc.py @@ -0,0 +1,163 @@ +"""Context wiring for the job-API IPC channel (logs only). + +The real IpcJobApiClient (threads + named pipes) is replaced by a recording fake +so these tests exercise the context's branching, not the transport — the +transport itself is covered by test_jobapi_client.py. The result stays on +output.json; only the logs move to IPC (and execution.log is suppressed). +""" + +import json +from pathlib import Path +from typing import Any + +import pytest + +from uipath.runtime.context import UiPathRuntimeContext +from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, PooledIpcSendLogHandler +from uipath.runtime.jobapi.pooled import set_pooled_log_sink +from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus + + +class _FakeIpcClient: + def __init__(self, endpoint: str, job_id: str, log: Any) -> None: + self.endpoint = endpoint + self.job_id = job_id + self.started = False + self.closed = False + self.logs: list[Any] = [] + + def start(self) -> None: + self.started = True + + def send_log(self, dto: Any) -> None: + self.logs.append(dto) + + def close(self) -> None: + self.closed = True + + +class _DummyInterceptor: + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.log_handler = kwargs.get("log_handler") + + def setup(self) -> None: + pass + + def teardown(self) -> None: + pass + + +@pytest.fixture(autouse=True) +def _patch(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("uipath.runtime.context.IpcJobApiClient", _FakeIpcClient) + monkeypatch.setattr( + "uipath.runtime.context.UiPathRuntimeLogsInterceptor", _DummyInterceptor + ) + + +def _ipc_ctx(tmp_path: Path, **kwargs: Any) -> UiPathRuntimeContext: + return UiPathRuntimeContext( + job_id="job-key", + runtime_dir=str(tmp_path / "rt"), + result_file="output.json", + ipc_endpoint="the-pipe", + ipc_job_id="job-guid", + **kwargs, + ) + + +def test_with_defaults_reads_ipc_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("UIPATH_JOB_API_IPC_ENDPOINT", "ep") + monkeypatch.setenv("UIPATH_JOB_ID", "jid") + + ctx = UiPathRuntimeContext.with_defaults() + + assert ctx.ipc_endpoint == "ep" + assert ctx.ipc_job_id == "jid" + assert ctx.ipc_active is True + + +def test_ipc_inactive_without_both_env_vars( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("UIPATH_JOB_API_IPC_ENDPOINT", "ep") + monkeypatch.delenv("UIPATH_JOB_ID", raising=False) + + ctx = UiPathRuntimeContext.with_defaults() + + assert ctx.ipc_active is False + + +def test_enter_starts_client_and_injects_ipc_log_handler(tmp_path: Path) -> None: + ctx = _ipc_ctx(tmp_path) + with ctx: + pass + + client = ctx.ipc_client + assert client.started is True + assert client.endpoint == "the-pipe" + assert client.job_id == "job-guid" + # The interceptor was handed our IPC handler (so no execution.log is opened). + assert isinstance(ctx.logs_interceptor.log_handler, IpcSendLogHandler) + + +def test_result_still_written_to_file_when_ipc_active(tmp_path: Path) -> None: + ctx = _ipc_ctx(tmp_path) + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} + ) + + # Only the logs move to IPC; the result stays on output.json for the handler. + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + assert content["output"] == {"foo": "bar"} + assert ctx.ipc_client.closed is True + + +def test_pooled_ipc_injects_pooled_handler_result_stays_on_file(tmp_path: Path) -> None: + # Pooled: a job id but no endpoint, plus a registered process-global sink. + calls: list[Any] = [] + set_pooled_log_sink(lambda job_id, log: calls.append((job_id, log))) + try: + ctx = UiPathRuntimeContext( + job_id="job-key", + runtime_dir=str(tmp_path / "rt"), + result_file="output.json", + ipc_job_id="job-guid", # no ipc_endpoint -> pooled path + ) + assert ctx.pooled_ipc_active is True + + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} + ) + + # The interceptor got the pooled handler (so execution.log is suppressed), and no per-job + # client was created (pooled forwards through the process-global sink instead). + assert isinstance(ctx.logs_interceptor.log_handler, PooledIpcSendLogHandler) + assert ctx.ipc_client is None + # The result still lands on output.json (logs-only channel). + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + assert content["output"] == {"foo": "bar"} + finally: + set_pooled_log_sink(None) + + +def test_no_ipc_log_handler_when_inactive(tmp_path: Path) -> None: + ctx = UiPathRuntimeContext( + job_id="job-key", + runtime_dir=str(tmp_path / "rt"), + result_file="output.json", + ) + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} + ) + + # No IPC → the interceptor builds its own (file) handler and no client is made. + assert ctx.logs_interceptor.log_handler is None + assert ctx.ipc_client is None diff --git a/tests/test_jobapi_client.py b/tests/test_jobapi_client.py new file mode 100644 index 0000000..47ff0be --- /dev/null +++ b/tests/test_jobapi_client.py @@ -0,0 +1,152 @@ +import asyncio +import logging +import time +import uuid + +import pytest + +import uipath.runtime.jobapi.client as client_mod +from uipath.runtime.jobapi.client import IpcJobApiClient +from uipath.runtime.jobapi.contract import IIpcLogSink, JobLogDto + + +def test_start_without_uipath_ipc_raises_loudly(monkeypatch): + # A missing transport must fail, not silently skip: the handler has already + # stopped reading the execution.log this client replaces. + monkeypatch.setattr( + "uipath.runtime.jobapi.client.importlib.util.find_spec", + lambda name: None, + ) + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + with pytest.raises(RuntimeError, match="uipath-ipc"): + client.start() + + +async def test_roundtrip_streams_logs_and_flushes_on_close(): + pytest.importorskip("uipath_ipc") + from uipath_ipc import IpcServer, NamedPipeServerTransport + + pipe = f"uipath-jobapi-test-{uuid.uuid4().hex}" + received: list[tuple[object, object]] = [] + + class _FakeHandler(IIpcLogSink): + async def SendLog(self, jobId, log): + received.append((jobId, log)) + + server = IpcServer( + transport=NamedPipeServerTransport(pipe), + services={IIpcLogSink: _FakeHandler()}, + request_timeout=None, + ) + async with server: + serve = asyncio.ensure_future(server.serve_forever()) + try: + job_id = str(uuid.uuid4()) + client = IpcJobApiClient(pipe, job_id, logging.getLogger("test")) + await asyncio.to_thread(client.start) + client.send_log(JobLogDto(Message="first", LogLevel=2)) + client.send_log(JobLogDto(Message="second", LogLevel=3)) + # close() flushes the queue before tearing the channel down. + await asyncio.to_thread(client.close) + finally: + serve.cancel() + + def _field(obj, name): + return getattr(obj, name) if hasattr(obj, name) else obj[name] + + assert [ + (_field(log, "Message"), _field(log, "LogLevel")) for _, log in received + ] == [ + ("first", 2), + ("second", 3), + ] + assert all(job == job_id for job, _ in received) + + +def test_start_times_out_when_thread_never_ready(monkeypatch): + monkeypatch.setattr( + "uipath.runtime.jobapi.client.importlib.util.find_spec", + lambda name: object(), + ) + monkeypatch.setattr(client_mod, "_STARTUP_TIMEOUT_S", 0.1) + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + monkeypatch.setattr(client, "_run", lambda: time.sleep(2)) + with pytest.raises(RuntimeError, match="did not connect"): + client.start() + + +def test_start_raises_when_setup_fails(monkeypatch): + monkeypatch.setattr( + "uipath.runtime.jobapi.client.importlib.util.find_spec", + lambda name: object(), + ) + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + + async def _boom(): + raise RuntimeError("setup boom") + + monkeypatch.setattr(client, "_setup", _boom) + with pytest.raises(RuntimeError, match="setup boom"): + client.start() + + +async def test_send_log_once_retries_then_drops(monkeypatch): + monkeypatch.setattr(client_mod, "_RETRY_DELAY_S", 0) + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + + class _Boom(IIpcLogSink): + async def SendLog(self, jobId, log): + raise RuntimeError("nope") + + client._proxy = _Boom() + assert await client._send_log_once(JobLogDto(Message="m")) is False + + +async def test_consume_mutes_after_repeated_failures(monkeypatch): + monkeypatch.setattr(client_mod, "_RETRY_DELAY_S", 0) + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + sends = 0 + + class _Boom(IIpcLogSink): + async def SendLog(self, jobId, log): + nonlocal sends + sends += 1 + raise RuntimeError("x") + + client._proxy = _Boom() + client._queue = asyncio.Queue() + task = asyncio.ensure_future(client._consume()) + + for i in range(client_mod._MAX_CONSECUTIVE_DROPS): + client._queue.put_nowait(JobLogDto(Message=str(i))) + await client._queue.join() + sends_while_trying = sends + + client._queue.put_nowait(JobLogDto(Message="muted")) + await client._queue.join() + assert sends == sends_while_trying # dropped while muted, no send attempted + + client._queue.put_nowait(client_mod._STOP) + await task + + +def test_send_log_noop_before_start(): + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + client.send_log(JobLogDto(Message="x")) + + +async def test_send_log_swallows_closed_loop_error(): + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + + class _DeadLoop: + def call_soon_threadsafe(self, *args): + raise RuntimeError("loop closed") + + client._loop = _DeadLoop() # type: ignore[assignment] + client._queue = asyncio.Queue() + client.send_log(JobLogDto(Message="x")) + + +def test_close_noop_before_start(): + client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) + client.close() diff --git a/tests/test_jobapi_contract.py b/tests/test_jobapi_contract.py new file mode 100644 index 0000000..e822ecd --- /dev/null +++ b/tests/test_jobapi_contract.py @@ -0,0 +1,41 @@ +import pytest + +from uipath.runtime.jobapi.contract import ( + IIpcLogSink, + JobLogDto, + LogLevel, +) + + +def test_endpoint_name_matches_dotnet_router_key(): + # uipath-ipc routes by the contract class __name__, which must equal the + # .NET interface name the Python executors register (IIpcLogSink, the log + # channel that .NET's IJobInvocationApi extends). + assert IIpcLogSink.__name__ == "IIpcLogSink" + + +def test_contract_only_declares_send_log(): + # Scope is logs only; the result stays on output.json. + assert hasattr(IIpcLogSink, "SendLog") + assert not hasattr(IIpcLogSink, "SetResult") + + +def test_log_level_values_match_microsoft_extensions(): + assert [ + LogLevel.TRACE, + LogLevel.DEBUG, + LogLevel.INFORMATION, + LogLevel.WARNING, + LogLevel.ERROR, + LogLevel.CRITICAL, + LogLevel.NONE, + ] == [0, 1, 2, 3, 4, 5, 6] + + +def test_job_log_dto_wire_shape(): + # The field names are the wire keys, so this pins them against the .NET DTO. + to_wire = pytest.importorskip("uipath_ipc").to_wire + assert to_wire(JobLogDto(Message="hi", LogLevel=3)) == { + "Message": "hi", + "LogLevel": 3, + } diff --git a/tests/test_jobapi_log_handler.py b/tests/test_jobapi_log_handler.py new file mode 100644 index 0000000..383ebf2 --- /dev/null +++ b/tests/test_jobapi_log_handler.py @@ -0,0 +1,57 @@ +import logging + +from uipath.runtime.jobapi.contract import JobLogDto, LogLevel +from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, _to_log_level + + +def test_to_log_level_mapping(): + assert _to_log_level(logging.CRITICAL) == LogLevel.CRITICAL + assert _to_log_level(logging.ERROR) == LogLevel.ERROR + assert _to_log_level(logging.WARNING) == LogLevel.WARNING + assert _to_log_level(logging.INFO) == LogLevel.INFORMATION + assert _to_log_level(logging.DEBUG) == LogLevel.DEBUG + # Below DEBUG collapses to TRACE; above CRITICAL clamps to CRITICAL. + assert _to_log_level(1) == LogLevel.TRACE + assert _to_log_level(logging.CRITICAL + 10) == LogLevel.CRITICAL + + +class _RecordingClient: + def __init__(self): + self.logs: list[JobLogDto] = [] + + def send_log(self, dto: JobLogDto) -> None: + self.logs.append(dto) + + +def test_emit_forwards_formatted_message_and_level(): + client = _RecordingClient() + handler = IpcSendLogHandler(client) # type: ignore[arg-type] + handler.setFormatter(logging.Formatter("%(message)s")) + + record = logging.LogRecord( + name="n", + level=logging.WARNING, + pathname="p", + lineno=1, + msg="hello %s", + args=("world",), + exc_info=None, + ) + handler.emit(record) + + assert len(client.logs) == 1 + assert client.logs[0].Message == "hello world" + assert client.logs[0].LogLevel == LogLevel.WARNING + + +def test_emit_swallows_client_errors(): + class _Boom: + def send_log(self, dto: JobLogDto) -> None: + raise RuntimeError("pipe down") + + handler = IpcSendLogHandler(_Boom()) # type: ignore[arg-type] + handler.setFormatter(logging.Formatter("%(message)s")) + record = logging.LogRecord("n", logging.INFO, "p", 1, "x", None, None) + + # handleError writes to sys.stderr but must not raise. + handler.emit(record) diff --git a/tests/test_jobapi_pooled.py b/tests/test_jobapi_pooled.py new file mode 100644 index 0000000..6785989 --- /dev/null +++ b/tests/test_jobapi_pooled.py @@ -0,0 +1,66 @@ +import logging +from typing import Any + +import pytest + +from uipath.runtime.jobapi.contract import JobLogDto, LogLevel +from uipath.runtime.jobapi.log_handler import PooledIpcSendLogHandler +from uipath.runtime.jobapi.pooled import get_pooled_log_sink, set_pooled_log_sink + + +@pytest.fixture(autouse=True) +def _clear_sink(): + # The sink is process-global; make sure a test never leaks it to the next. + set_pooled_log_sink(None) + yield + set_pooled_log_sink(None) + + +def test_sink_registry_set_and_clear(): + assert get_pooled_log_sink() is None + + def sink(job_id: str, log: JobLogDto) -> None: + pass + + set_pooled_log_sink(sink) + assert get_pooled_log_sink() is sink + + set_pooled_log_sink(None) + assert get_pooled_log_sink() is None + + +def test_pooled_handler_forwards_tagged_with_job_id(): + calls: list[tuple[str, Any]] = [] + handler = PooledIpcSendLogHandler( + "job-key-1", lambda jid, log: calls.append((jid, log)) + ) + handler.setFormatter(logging.Formatter("%(message)s")) + + record = logging.LogRecord( + name="n", + level=logging.WARNING, + pathname="p", + lineno=1, + msg="hello %s", + args=("world",), + exc_info=None, + ) + handler.emit(record) + + assert len(calls) == 1 + job_id, log = calls[0] + assert job_id == "job-key-1" + assert log.Message == "hello world" + assert log.LogLevel == LogLevel.WARNING + + +def test_pooled_handler_swallows_sink_errors(): + def boom(job_id: str, log: JobLogDto) -> None: + raise RuntimeError("pipe down") + + handler = PooledIpcSendLogHandler("job-key-1", boom) + handler.setFormatter(logging.Formatter("%(message)s")) + record = logging.LogRecord("n", logging.INFO, "p", 1, "x", None, None) + + # handleError writes to stderr but must not raise. + handler.emit(record) diff --git a/uv.lock b/uv.lock index a0dd246..6576d81 100644 --- a/uv.lock +++ b/uv.lock @@ -7,6 +7,7 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P2D" [options.exclude-newer-package] +uipath-ipc = false uipath-core = false [[package]] @@ -1151,9 +1152,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/37/47a4e12bc7bdcfab4bd4e8e1f2a5c083bd59fca42fd431026a3f14c642a3/uipath_core-0.5.31-py3-none-any.whl", hash = "sha256:3dff5ecb236bf46af683c34d79bb2cf458a942ec041e1cbf7b4825ae246ff00c", size = 55040, upload-time = "2026-07-15T06:56:02.446Z" }, ] +[[package]] +name = "uipath-ipc" +version = "2.5.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/2c/6b/53d9725d6abd1dab447300a7f999332f530678e3fabd38fc31171a5a9a6f/uipath_ipc-2.5.2.tar.gz", hash = "sha256:d69c3d7c1ad1a25ef7f9f1d78c505f5d2ed7daa7227bb202404fa3d47e0ba12e", size = 86360, upload-time = "2026-07-24T09:44:35.159Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/29/eb/6def505a0d351119da27b342c11eb0767a31a7f100022d35e349c5194eb3/uipath_ipc-2.5.2-py3-none-any.whl", hash = "sha256:617f25f35377d87956165a875b0966a3cd69ae6724fcd533c93a7a6a9460fc4f", size = 50345, upload-time = "2026-07-24T09:44:33.7Z" }, +] + [[package]] name = "uipath-runtime" -version = "0.13.4" +version = "0.13.5" source = { editable = "." } dependencies = [ { name = "chardet" }, @@ -1161,6 +1171,11 @@ dependencies = [ { name = "vadersentiment" }, ] +[package.optional-dependencies] +ipc = [ + { name = "uipath-ipc" }, +] + [package.dev-dependencies] dev = [ { name = "bandit" }, @@ -1174,14 +1189,17 @@ dev = [ { name = "pytest-trio" }, { name = "ruff" }, { name = "rust-just" }, + { name = "uipath-ipc" }, ] [package.metadata] requires-dist = [ { name = "chardet", specifier = ">=5.2.0,<8.0" }, { name = "uipath-core", specifier = ">=0.5.31,<0.6.0" }, + { name = "uipath-ipc", marker = "extra == 'ipc'", specifier = ">=2.5.1,<2.6.0" }, { name = "vadersentiment", specifier = ">=3.3.2,<4.0" }, ] +provides-extras = ["ipc"] [package.metadata.requires-dev] dev = [ @@ -1196,6 +1214,7 @@ dev = [ { name = "pytest-trio", specifier = ">=0.8.0" }, { name = "ruff", specifier = ">=0.9.4" }, { name = "rust-just", specifier = ">=1.39.0" }, + { name = "uipath-ipc", specifier = ">=2.5.1,<2.6.0" }, ] [[package]] From 88f6cd33d147aa907ef9b3f0951c46f113e09e8f Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Tue, 8 Sep 2026 14:02:16 +0200 Subject: [PATCH 2/2] feat(runtime): in-memory output sinks for host-driven log/result delivery Replace the uipath-ipc-coupled jobapi client with transport-agnostic in-memory sinks (output_sinks.py): the host (uipath-python) installs a log handler and a result sink, and uipath-runtime keeps no IPC dependency. Drops the [ipc] extra and the jobapi package. Review fixes: - context.__exit__ isolates the result sink so a delivery failure can't clobber the already-persisted output.json as a FAULTED shutdown error - skip the duplicate .args write when split_output_arguments and a sink are both on - cover that a host-provided (unowned) log handler is not closed on teardown Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 6 - src/uipath/runtime/context.py | 87 +++++------ src/uipath/runtime/jobapi/__init__.py | 29 ---- src/uipath/runtime/jobapi/client.py | 184 ----------------------- src/uipath/runtime/jobapi/contract.py | 39 ----- src/uipath/runtime/jobapi/log_handler.py | 61 -------- src/uipath/runtime/jobapi/pooled.py | 28 ---- src/uipath/runtime/output_sinks.py | 41 +++++ tests/test_context_ipc.py | 163 -------------------- tests/test_interceptor.py | 39 +++++ tests/test_jobapi_client.py | 152 ------------------- tests/test_jobapi_contract.py | 41 ----- tests/test_jobapi_log_handler.py | 57 ------- tests/test_jobapi_pooled.py | 66 -------- tests/test_output_sinks.py | 184 +++++++++++++++++++++++ uv.lock | 19 --- 16 files changed, 298 insertions(+), 898 deletions(-) delete mode 100644 src/uipath/runtime/jobapi/__init__.py delete mode 100644 src/uipath/runtime/jobapi/client.py delete mode 100644 src/uipath/runtime/jobapi/contract.py delete mode 100644 src/uipath/runtime/jobapi/log_handler.py delete mode 100644 src/uipath/runtime/jobapi/pooled.py create mode 100644 src/uipath/runtime/output_sinks.py delete mode 100644 tests/test_context_ipc.py delete mode 100644 tests/test_jobapi_client.py delete mode 100644 tests/test_jobapi_contract.py delete mode 100644 tests/test_jobapi_log_handler.py delete mode 100644 tests/test_jobapi_pooled.py create mode 100644 tests/test_output_sinks.py diff --git a/pyproject.toml b/pyproject.toml index 95c926d..ebe7206 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,10 +20,6 @@ maintainers = [ { name = "Cristian Pufu", email = "cristian.pufu@uipath.com" }, ] -[project.optional-dependencies] -# Job-API IPC log channel; required when UIPATH_JOB_API_IPC_ENDPOINT is set. -ipc = ["uipath-ipc>=2.5.1, <2.6.0"] - [project.urls] Homepage = "https://uipath.com" Repository = "https://github.com/UiPath/uipath-runtime-python" @@ -46,7 +42,6 @@ dev = [ "pytest-cov>=4.1.0", "pytest-mock>=3.11.1", "pre-commit>=4.1.0", - "uipath-ipc>=2.5.1, <2.6.0", ] [tool.hatch.build.targets.wheel] @@ -128,7 +123,6 @@ exclude-newer = "2 days" [tool.uv.exclude-newer-package] uipath-core = false -uipath-ipc = false [[tool.uv.index]] name = "testpypi" diff --git a/src/uipath/runtime/context.py b/src/uipath/runtime/context.py index 91f8fb9..ad79866 100644 --- a/src/uipath/runtime/context.py +++ b/src/uipath/runtime/context.py @@ -18,10 +18,8 @@ UiPathErrorContract, UiPathRuntimeError, ) -from uipath.runtime.jobapi.client import IpcJobApiClient -from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, PooledIpcSendLogHandler -from uipath.runtime.jobapi.pooled import get_pooled_log_sink from uipath.runtime.logging._interceptor import UiPathRuntimeLogsInterceptor +from uipath.runtime.output_sinks import ResultSink, get_log_handler, get_result_sink from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus logger = logging.getLogger(__name__) @@ -123,38 +121,8 @@ class UiPathRuntimeContext(BaseModel): keep_state_file: bool = Field( False, description="Prevents deletion of state file before running." ) - ipc_endpoint: str | None = Field( - None, - description=( - "uipath-ipc endpoint (UIPATH_JOB_API_IPC_ENDPOINT) for streaming logs to " - "the handler in place of execution.log. The result stays on output.json." - ), - ) - ipc_job_id: str | None = Field( - None, - description="Handler job id (UIPATH_JOB_ID) used to route the IPC calls.", - ) - ipc_client: Any = Field(default=None, exclude=True, repr=False) - model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") - @property - def ipc_active(self) -> bool: - """Whether logs flow over a per-job IPC pipe (non-pooled) instead of to files.""" - return bool(self.ipc_endpoint and self.ipc_job_id) - - @property - def pooled_ipc_active(self) -> bool: - """Whether logs stream over the pooled server's callback rather than a per-job pipe. - - True when there is a job id but no endpoint, and the pooled server registered its sink. - """ - return ( - bool(self.ipc_job_id) - and not self.ipc_endpoint - and get_pooled_log_sink() is not None - ) - def _apply_execution_source(self) -> None: """Derive execution_source from the command, if not already set. @@ -270,22 +238,9 @@ def __enter__(self): Returns: The runtime context instance """ - # Intercept all stdout/stderr/logs - # Write to file (runtime), stdout (debug) or log handler (if provided) - log_handler: logging.Handler | None = None - if self.ipc_active: - assert self.ipc_endpoint is not None and self.ipc_job_id is not None - self.ipc_client = IpcJobApiClient( - self.ipc_endpoint, self.ipc_job_id, logger - ) - self.ipc_client.start() - log_handler = IpcSendLogHandler(self.ipc_client) - log_handler.setFormatter(logging.Formatter("%(message)s")) - elif self.pooled_ipc_active: - sink = get_pooled_log_sink() - assert self.ipc_job_id is not None and sink is not None - log_handler = PooledIpcSendLogHandler(self.ipc_job_id, sink) - log_handler.setFormatter(logging.Formatter("%(message)s")) + # Intercept all stdout/stderr/logs. A host may have installed a log handler (routing records + # to IPC etc.); otherwise the interceptor opens execution.log. + log_handler = get_log_handler() self.logs_interceptor = UiPathRuntimeLogsInterceptor( min_level=self.logs_min_level, @@ -359,6 +314,21 @@ def __exit__(self, exc_type, exc_val, exc_tb): with open(self.output_file, "w") as f: json.dump(output_payload, f, default=str) + # Hand the result to a host-installed sink, if any. Successful/Faulted only — Suspended + # keeps output.json (resume triggers live there); output.json is written above regardless. + # Best-effort and ISOLATED: the sink is a side channel (e.g. IPC delivery), so a failure + # here must not reach the catch-all below — that would rewrite the already-persisted good + # output.json as FAULTED and fault a job that actually succeeded. + result_sink = get_result_sink() + if result_sink is not None and self.result.status in ( + UiPathRuntimeStatus.SUCCESSFUL, + UiPathRuntimeStatus.FAULTED, + ): + try: + self._deliver_result(result_sink, output_payload) + except Exception as sink_error: + logger.error(f"Failed to deliver result to sink: {str(sink_error)}") + # Don't suppress exceptions return False @@ -402,8 +372,21 @@ def __exit__(self, exc_type, exc_val, exc_tb): # Restore original logging if hasattr(self, "logs_interceptor"): self.logs_interceptor.teardown() - if self.ipc_client is not None: - self.ipc_client.close() + + def _deliver_result(self, sink: "ResultSink", output_payload: Any) -> None: + """Spill the output arguments to a file, then hand the result + that path to the host's sink. + + The unbounded output arguments never travel inline — they go to a file the host delivers (it + can stream it straight on, off-heap). The host maps the runtime result to its own wire shape. + """ + args_path = self.resolved_output_arguments_file_path + # The split-output-arguments branch may already have spilled this exact payload to this exact + # path; don't serialize the (potentially large) payload to disk a second time. + if not (self.split_output_arguments and self.job_id): + os.makedirs(os.path.dirname(args_path), exist_ok=True) + with open(args_path, "w") as f: + json.dump(output_payload, f, default=str) + sink(self.result, args_path) @cached_property def resolved_result_file_path(self) -> str: @@ -468,8 +451,6 @@ def with_defaults( base.tenant_id = os.environ.get("UIPATH_TENANT_ID") base.process_key = os.environ.get("UIPATH_PROCESS_UUID") base.folder_key = os.environ.get("UIPATH_FOLDER_KEY") - base.ipc_endpoint = os.environ.get("UIPATH_JOB_API_IPC_ENDPOINT") - base.ipc_job_id = os.environ.get("UIPATH_JOB_ID") # Override with kwargs for k, v in kwargs.items(): diff --git a/src/uipath/runtime/jobapi/__init__.py b/src/uipath/runtime/jobapi/__init__.py deleted file mode 100644 index 8945ac2..0000000 --- a/src/uipath/runtime/jobapi/__init__.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Job-API IPC: stream job logs back to the handler (per-job client or pooled callback).""" - -from uipath.runtime.jobapi.client import IpcJobApiClient -from uipath.runtime.jobapi.contract import ( - IIpcLogSink, - JobLogDto, - LogLevel, -) -from uipath.runtime.jobapi.log_handler import ( - IpcSendLogHandler, - PooledIpcSendLogHandler, -) -from uipath.runtime.jobapi.pooled import ( - PooledLogSink, - get_pooled_log_sink, - set_pooled_log_sink, -) - -__all__ = [ - "IIpcLogSink", - "IpcJobApiClient", - "IpcSendLogHandler", - "JobLogDto", - "LogLevel", - "PooledIpcSendLogHandler", - "PooledLogSink", - "get_pooled_log_sink", - "set_pooled_log_sink", -] diff --git a/src/uipath/runtime/jobapi/client.py b/src/uipath/runtime/jobapi/client.py deleted file mode 100644 index e117ec7..0000000 --- a/src/uipath/runtime/jobapi/client.py +++ /dev/null @@ -1,184 +0,0 @@ -"""uipath-ipc client that streams job logs to the handler (logs only).""" - -from __future__ import annotations - -import asyncio -import importlib.util -import logging -import sys -import threading -import time -from typing import Any - -from uipath.runtime.jobapi.contract import IIpcLogSink, JobLogDto - -_SEND_TIMEOUT_S = 1.0 -_MAX_SEND_ATTEMPTS = 3 -_RETRY_DELAY_S = 0.2 -_MAX_CONSECUTIVE_DROPS = 3 -_FAILURE_COOLDOWN_S = 30.0 -_STARTUP_TIMEOUT_S = 10.0 -_SHUTDOWN_TIMEOUT_S = 15.0 - -_STOP = object() - - -class IpcJobApiClient: - """Owns a uipath-ipc client on a private event-loop thread. - - The runtime's ``__exit__`` is synchronous but IPC is asyncio-based, so the - client runs its loop on its own thread (a Proactor loop on Windows, which - named pipes require). Logs are enqueued without blocking the caller and - drained one at a time, FIFO; the queue is flushed at job end. A send that - fails is retried a few times, then that one entry is dropped, and after - repeated failures forwarding is muted for a cooldown — mirroring the JS - coded-functions log forwarder. - """ - - def __init__(self, endpoint: str, job_id: str, log: logging.Logger) -> None: - """Configure the client; call ``start()`` to connect.""" - self._endpoint = endpoint - self._job_id = job_id - self._log = log - self._loop: asyncio.AbstractEventLoop | None = None - self._thread: threading.Thread | None = None - self._queue: asyncio.Queue[Any] | None = None - self._proxy: IIpcLogSink | None = None - self._client: Any = None - self._consumer: asyncio.Task[None] | None = None - self._ready = threading.Event() - self._start_error: BaseException | None = None - self._stopping = False - - def start(self) -> None: - """Start the loop thread and connect; raise if that fails or times out.""" - if importlib.util.find_spec("uipath_ipc") is None: - raise RuntimeError( - "Job-API IPC was requested (UIPATH_JOB_API_IPC_ENDPOINT is set) but " - "the 'uipath-ipc' package is not installed. Install it (e.g. " - "'pip install uipath-ipc') to stream logs over IPC." - ) - - self._thread = threading.Thread( - target=self._run, name="uipath-jobapi-ipc", daemon=True - ) - self._thread.start() - if not self._ready.wait(timeout=_STARTUP_TIMEOUT_S): - raise RuntimeError( - f"Job-API IPC client did not connect within {_STARTUP_TIMEOUT_S}s " - f"(endpoint {self._endpoint!r})." - ) - if self._start_error is not None: - raise self._start_error - - def _run(self) -> None: - try: - if sys.platform == "win32": - self._loop = asyncio.ProactorEventLoop() - else: - self._loop = asyncio.new_event_loop() - asyncio.set_event_loop(self._loop) - self._loop.run_until_complete(self._setup()) - except BaseException as e: - self._start_error = e - self._ready.set() - return - self._ready.set() - try: - self._loop.run_forever() - finally: - self._loop.close() - - async def _setup(self) -> None: - from uipath_ipc import IpcClient, NamedPipeClientTransport - - self._client = IpcClient( - transport=NamedPipeClientTransport(self._endpoint), - request_timeout=None, - ) - self._proxy = self._client.get_proxy(IIpcLogSink) - self._queue = asyncio.Queue() - self._consumer = asyncio.ensure_future(self._consume()) - - async def _consume(self) -> None: - assert self._queue is not None - consecutive_drops = 0 - muted_until = 0.0 - while True: - item = await self._queue.get() - try: - if item is _STOP: - return - if time.monotonic() < muted_until: - continue - if await self._send_log_once(item): - consecutive_drops = 0 - else: - consecutive_drops += 1 - if consecutive_drops >= _MAX_CONSECUTIVE_DROPS: - muted_until = time.monotonic() + _FAILURE_COOLDOWN_S - consecutive_drops = 0 - self._log.warning( - "Job-API IPC: log forwarding paused for %ss after " - "repeated send failures.", - int(_FAILURE_COOLDOWN_S), - ) - finally: - self._queue.task_done() - - async def _send_log_once(self, dto: JobLogDto) -> bool: - assert self._proxy is not None - last_err: BaseException | None = None - for attempt in range(1, _MAX_SEND_ATTEMPTS + 1): - try: - await asyncio.wait_for( - self._proxy.SendLog(self._job_id, dto), timeout=_SEND_TIMEOUT_S - ) - return True - except Exception as e: - last_err = e - if attempt < _MAX_SEND_ATTEMPTS: - await asyncio.sleep(_RETRY_DELAY_S) - self._log.debug("Job-API IPC: dropped one log entry: %s", last_err) - return False - - def send_log(self, dto: JobLogDto) -> None: - """Enqueue one log entry for sending (thread-safe, never blocks).""" - loop = self._loop - queue = self._queue - if loop is None or queue is None or self._stopping: - return - try: - loop.call_soon_threadsafe(queue.put_nowait, dto) - except RuntimeError: - pass - - def close(self) -> None: - """Flush queued logs, close the client, and join the loop thread.""" - self._stopping = True - loop = self._loop - thread = self._thread - if loop is None or thread is None or not thread.is_alive(): - return - try: - fut = asyncio.run_coroutine_threadsafe(self._shutdown(), loop) - fut.result(timeout=_SHUTDOWN_TIMEOUT_S) - except Exception as e: - self._log.debug("Job-API IPC: error during shutdown: %s", e) - finally: - loop.call_soon_threadsafe(loop.stop) - thread.join(timeout=_SHUTDOWN_TIMEOUT_S) - - async def _shutdown(self) -> None: - if self._queue is not None and self._consumer is not None: - await self._queue.join() - self._queue.put_nowait(_STOP) - try: - await asyncio.wait_for(self._consumer, timeout=_SHUTDOWN_TIMEOUT_S) - except Exception: - self._consumer.cancel() - if self._client is not None: - try: - await self._client.aclose() - except Exception as e: - self._log.debug("Job-API IPC: error closing client: %s", e) diff --git a/src/uipath/runtime/jobapi/contract.py b/src/uipath/runtime/jobapi/contract.py deleted file mode 100644 index 59a44b1..0000000 --- a/src/uipath/runtime/jobapi/contract.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Python mirror of the .NET ``IIpcLogSink`` CoreIpc contract (logs only). - -uipath-ipc routes by the contract class ``__name__`` and method name and serializes each -argument by its declared field names, so the class name, the method name, and every field -below must match the .NET side exactly. The result is not sent over IPC — it stays on -``output.json``, which the handler still reads. -""" - -from abc import ABC, abstractmethod -from dataclasses import dataclass -from enum import IntEnum - - -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 - - -@dataclass -class JobLogDto: - """A single log entry (PascalCase fields to match the wire).""" - - Message: str = "" - LogLevel: int = LogLevel.INFORMATION.value - - -class IIpcLogSink(ABC): - """The handler's log-sink contract; the class name is the CoreIpc endpoint key.""" - - @abstractmethod - async def SendLog(self, jobId: str, log: JobLogDto) -> None: - """Forward one log entry (one-way).""" diff --git a/src/uipath/runtime/jobapi/log_handler.py b/src/uipath/runtime/jobapi/log_handler.py deleted file mode 100644 index a7642bd..0000000 --- a/src/uipath/runtime/jobapi/log_handler.py +++ /dev/null @@ -1,61 +0,0 @@ -"""Logging handlers that forward records to the handler over uipath-ipc.""" - -import logging - -from uipath.runtime.jobapi.client import IpcJobApiClient -from uipath.runtime.jobapi.contract import JobLogDto, LogLevel -from uipath.runtime.jobapi.pooled import PooledLogSink - - -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 - - -class IpcSendLogHandler(logging.Handler): - """Forwards each record to the handler's IIpcLogSink via the non-pooled per-job client.""" - - def __init__(self, client: IpcJobApiClient) -> None: - """Wrap the IPC client the records are forwarded through.""" - super().__init__() - self._client = client - - def emit(self, record: logging.LogRecord) -> None: - """Format the record and hand it to the IPC client (non-blocking).""" - try: - message = self.format(record) - self._client.send_log( - JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno)) - ) - except Exception: - self.handleError(record) - - -class PooledIpcSendLogHandler(logging.Handler): - """Forwards each record to the pooled process-global sink, tagged with the job id.""" - - def __init__(self, job_id: str, sink: PooledLogSink) -> None: - """Bind the job id every record is tagged with and the sink to forward through.""" - super().__init__() - self._job_id = job_id - self._sink = sink - - def emit(self, record: logging.LogRecord) -> None: - """Format the record and hand it to the pooled sink (non-blocking).""" - try: - message = self.format(record) - self._sink( - self._job_id, - JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno)), - ) - except Exception: - self.handleError(record) diff --git a/src/uipath/runtime/jobapi/pooled.py b/src/uipath/runtime/jobapi/pooled.py deleted file mode 100644 index 9a61740..0000000 --- a/src/uipath/runtime/jobapi/pooled.py +++ /dev/null @@ -1,28 +0,0 @@ -"""Process-global log sink for the pooled path. - -In pooled mode the job runs in-process inside ``uipath server``, which owns the CoreIpc -callback to the handler. The runtime can't reach that callback directly, so the pooled -server registers a sink here and the runtime's log handler forwards to it. The sink is -called from the job's logging (a worker thread), so it must be non-blocking and thread-safe. -""" - -from __future__ import annotations - -from typing import Callable - -from uipath.runtime.jobapi.contract import JobLogDto - -PooledLogSink = Callable[[str, JobLogDto], None] - -_sink: "PooledLogSink | None" = None - - -def set_pooled_log_sink(sink: "PooledLogSink | None") -> None: - """Register (or clear, with ``None``) the process-global pooled log sink.""" - global _sink - _sink = sink - - -def get_pooled_log_sink() -> "PooledLogSink | None": - """The registered pooled log sink, or None when not in a pooled server.""" - return _sink diff --git a/src/uipath/runtime/output_sinks.py b/src/uipath/runtime/output_sinks.py new file mode 100644 index 0000000..ddeedc4 --- /dev/null +++ b/src/uipath/runtime/output_sinks.py @@ -0,0 +1,41 @@ +"""In-memory sinks the host installs to receive a job's logs and result. + +By default uipath-runtime writes the logs to ``execution.log`` and the result to ``output.json``. +A host that drives the runtime in-process (uipath-python's ``uipath run`` / ``uipath server``) can +instead install a log handler and a result sink here — set before a job runs, cleared after — and +the runtime routes to them. Transport-agnostic: the host decides IPC/HTTP/whatever. The runtime +keeps no such dependency; these are plain in-process callables. +""" + +from __future__ import annotations + +import logging +from typing import Any, Callable + +# (result, output_arguments_file_path) -> None. Called once, at job end, for Successful/Faulted. +ResultSink = Callable[[Any, str], None] + +_log_handler: "logging.Handler | None" = None +_result_sink: "ResultSink | None" = None + + +def set_log_handler(handler: "logging.Handler | None") -> None: + """Install (or clear, with ``None``) the log handler the runtime routes records to.""" + global _log_handler + _log_handler = handler + + +def get_log_handler() -> "logging.Handler | None": + """The installed log handler, or None to keep the file (execution.log).""" + return _log_handler + + +def set_result_sink(sink: "ResultSink | None") -> None: + """Install (or clear, with ``None``) the sink the runtime hands the final result to.""" + global _result_sink + _result_sink = sink + + +def get_result_sink() -> "ResultSink | None": + """The installed result sink, or None to keep the file (output.json).""" + return _result_sink diff --git a/tests/test_context_ipc.py b/tests/test_context_ipc.py deleted file mode 100644 index 77eacfe..0000000 --- a/tests/test_context_ipc.py +++ /dev/null @@ -1,163 +0,0 @@ -"""Context wiring for the job-API IPC channel (logs only). - -The real IpcJobApiClient (threads + named pipes) is replaced by a recording fake -so these tests exercise the context's branching, not the transport — the -transport itself is covered by test_jobapi_client.py. The result stays on -output.json; only the logs move to IPC (and execution.log is suppressed). -""" - -import json -from pathlib import Path -from typing import Any - -import pytest - -from uipath.runtime.context import UiPathRuntimeContext -from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, PooledIpcSendLogHandler -from uipath.runtime.jobapi.pooled import set_pooled_log_sink -from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus - - -class _FakeIpcClient: - def __init__(self, endpoint: str, job_id: str, log: Any) -> None: - self.endpoint = endpoint - self.job_id = job_id - self.started = False - self.closed = False - self.logs: list[Any] = [] - - def start(self) -> None: - self.started = True - - def send_log(self, dto: Any) -> None: - self.logs.append(dto) - - def close(self) -> None: - self.closed = True - - -class _DummyInterceptor: - def __init__(self, *args: Any, **kwargs: Any) -> None: - self.log_handler = kwargs.get("log_handler") - - def setup(self) -> None: - pass - - def teardown(self) -> None: - pass - - -@pytest.fixture(autouse=True) -def _patch(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr("uipath.runtime.context.IpcJobApiClient", _FakeIpcClient) - monkeypatch.setattr( - "uipath.runtime.context.UiPathRuntimeLogsInterceptor", _DummyInterceptor - ) - - -def _ipc_ctx(tmp_path: Path, **kwargs: Any) -> UiPathRuntimeContext: - return UiPathRuntimeContext( - job_id="job-key", - runtime_dir=str(tmp_path / "rt"), - result_file="output.json", - ipc_endpoint="the-pipe", - ipc_job_id="job-guid", - **kwargs, - ) - - -def test_with_defaults_reads_ipc_env( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.chdir(tmp_path) - monkeypatch.setenv("UIPATH_JOB_API_IPC_ENDPOINT", "ep") - monkeypatch.setenv("UIPATH_JOB_ID", "jid") - - ctx = UiPathRuntimeContext.with_defaults() - - assert ctx.ipc_endpoint == "ep" - assert ctx.ipc_job_id == "jid" - assert ctx.ipc_active is True - - -def test_ipc_inactive_without_both_env_vars( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.chdir(tmp_path) - monkeypatch.setenv("UIPATH_JOB_API_IPC_ENDPOINT", "ep") - monkeypatch.delenv("UIPATH_JOB_ID", raising=False) - - ctx = UiPathRuntimeContext.with_defaults() - - assert ctx.ipc_active is False - - -def test_enter_starts_client_and_injects_ipc_log_handler(tmp_path: Path) -> None: - ctx = _ipc_ctx(tmp_path) - with ctx: - pass - - client = ctx.ipc_client - assert client.started is True - assert client.endpoint == "the-pipe" - assert client.job_id == "job-guid" - # The interceptor was handed our IPC handler (so no execution.log is opened). - assert isinstance(ctx.logs_interceptor.log_handler, IpcSendLogHandler) - - -def test_result_still_written_to_file_when_ipc_active(tmp_path: Path) -> None: - ctx = _ipc_ctx(tmp_path) - with ctx: - ctx.result = UiPathRuntimeResult( - status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} - ) - - # Only the logs move to IPC; the result stays on output.json for the handler. - content = json.loads(Path(ctx.resolved_result_file_path).read_text()) - assert content["output"] == {"foo": "bar"} - assert ctx.ipc_client.closed is True - - -def test_pooled_ipc_injects_pooled_handler_result_stays_on_file(tmp_path: Path) -> None: - # Pooled: a job id but no endpoint, plus a registered process-global sink. - calls: list[Any] = [] - set_pooled_log_sink(lambda job_id, log: calls.append((job_id, log))) - try: - ctx = UiPathRuntimeContext( - job_id="job-key", - runtime_dir=str(tmp_path / "rt"), - result_file="output.json", - ipc_job_id="job-guid", # no ipc_endpoint -> pooled path - ) - assert ctx.pooled_ipc_active is True - - with ctx: - ctx.result = UiPathRuntimeResult( - status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} - ) - - # The interceptor got the pooled handler (so execution.log is suppressed), and no per-job - # client was created (pooled forwards through the process-global sink instead). - assert isinstance(ctx.logs_interceptor.log_handler, PooledIpcSendLogHandler) - assert ctx.ipc_client is None - # The result still lands on output.json (logs-only channel). - content = json.loads(Path(ctx.resolved_result_file_path).read_text()) - assert content["output"] == {"foo": "bar"} - finally: - set_pooled_log_sink(None) - - -def test_no_ipc_log_handler_when_inactive(tmp_path: Path) -> None: - ctx = UiPathRuntimeContext( - job_id="job-key", - runtime_dir=str(tmp_path / "rt"), - result_file="output.json", - ) - with ctx: - ctx.result = UiPathRuntimeResult( - status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} - ) - - # No IPC → the interceptor builds its own (file) handler and no client is made. - assert ctx.logs_interceptor.log_handler is None - assert ctx.ipc_client is None diff --git a/tests/test_interceptor.py b/tests/test_interceptor.py index b69e093..e8c50ef 100644 --- a/tests/test_interceptor.py +++ b/tests/test_interceptor.py @@ -153,6 +153,45 @@ def tracked_close(): assert call_order.index("detach") < call_order.index("handler_close") +class TestInterceptorWithHostHandler: + """A host-provided log_handler is USED but not OWNED: records reach it, teardown must not close it. + + This is the load-bearing invariant for the IPC output path — the host installs a handler that + forwards records to its own sink and reuses it across jobs, so the interceptor closing it would + break delivery. + """ + + def test_host_handler_receives_records_and_survives_teardown(self): + records: list[logging.LogRecord] = [] + closed = {"value": False} + + class _HostHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + def close(self) -> None: + closed["value"] = True + super().close() + + handler = _HostHandler() + interceptor = UiPathRuntimeLogsInterceptor( + min_level="INFO", job_id="job-1", log_handler=handler + ) + # A host-provided handler is not owned by the interceptor. + assert interceptor._owns_handler is False + + interceptor.setup() + try: + logging.getLogger("runtime").info("hello from the job") + finally: + interceptor.teardown() + + # setup() attached the host handler and a record flowed through it... + assert any(r.getMessage() == "hello from the job" for r in records) + # ...and teardown did NOT close a handler it does not own. + assert closed["value"] is False + + class TestInterceptorWithJobId: """When job_id is set, a file handler is used — no utf8_stdout wrapper.""" diff --git a/tests/test_jobapi_client.py b/tests/test_jobapi_client.py deleted file mode 100644 index 47ff0be..0000000 --- a/tests/test_jobapi_client.py +++ /dev/null @@ -1,152 +0,0 @@ -import asyncio -import logging -import time -import uuid - -import pytest - -import uipath.runtime.jobapi.client as client_mod -from uipath.runtime.jobapi.client import IpcJobApiClient -from uipath.runtime.jobapi.contract import IIpcLogSink, JobLogDto - - -def test_start_without_uipath_ipc_raises_loudly(monkeypatch): - # A missing transport must fail, not silently skip: the handler has already - # stopped reading the execution.log this client replaces. - monkeypatch.setattr( - "uipath.runtime.jobapi.client.importlib.util.find_spec", - lambda name: None, - ) - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - with pytest.raises(RuntimeError, match="uipath-ipc"): - client.start() - - -async def test_roundtrip_streams_logs_and_flushes_on_close(): - pytest.importorskip("uipath_ipc") - from uipath_ipc import IpcServer, NamedPipeServerTransport - - pipe = f"uipath-jobapi-test-{uuid.uuid4().hex}" - received: list[tuple[object, object]] = [] - - class _FakeHandler(IIpcLogSink): - async def SendLog(self, jobId, log): - received.append((jobId, log)) - - server = IpcServer( - transport=NamedPipeServerTransport(pipe), - services={IIpcLogSink: _FakeHandler()}, - request_timeout=None, - ) - async with server: - serve = asyncio.ensure_future(server.serve_forever()) - try: - job_id = str(uuid.uuid4()) - client = IpcJobApiClient(pipe, job_id, logging.getLogger("test")) - await asyncio.to_thread(client.start) - client.send_log(JobLogDto(Message="first", LogLevel=2)) - client.send_log(JobLogDto(Message="second", LogLevel=3)) - # close() flushes the queue before tearing the channel down. - await asyncio.to_thread(client.close) - finally: - serve.cancel() - - def _field(obj, name): - return getattr(obj, name) if hasattr(obj, name) else obj[name] - - assert [ - (_field(log, "Message"), _field(log, "LogLevel")) for _, log in received - ] == [ - ("first", 2), - ("second", 3), - ] - assert all(job == job_id for job, _ in received) - - -def test_start_times_out_when_thread_never_ready(monkeypatch): - monkeypatch.setattr( - "uipath.runtime.jobapi.client.importlib.util.find_spec", - lambda name: object(), - ) - monkeypatch.setattr(client_mod, "_STARTUP_TIMEOUT_S", 0.1) - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - monkeypatch.setattr(client, "_run", lambda: time.sleep(2)) - with pytest.raises(RuntimeError, match="did not connect"): - client.start() - - -def test_start_raises_when_setup_fails(monkeypatch): - monkeypatch.setattr( - "uipath.runtime.jobapi.client.importlib.util.find_spec", - lambda name: object(), - ) - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - - async def _boom(): - raise RuntimeError("setup boom") - - monkeypatch.setattr(client, "_setup", _boom) - with pytest.raises(RuntimeError, match="setup boom"): - client.start() - - -async def test_send_log_once_retries_then_drops(monkeypatch): - monkeypatch.setattr(client_mod, "_RETRY_DELAY_S", 0) - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - - class _Boom(IIpcLogSink): - async def SendLog(self, jobId, log): - raise RuntimeError("nope") - - client._proxy = _Boom() - assert await client._send_log_once(JobLogDto(Message="m")) is False - - -async def test_consume_mutes_after_repeated_failures(monkeypatch): - monkeypatch.setattr(client_mod, "_RETRY_DELAY_S", 0) - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - sends = 0 - - class _Boom(IIpcLogSink): - async def SendLog(self, jobId, log): - nonlocal sends - sends += 1 - raise RuntimeError("x") - - client._proxy = _Boom() - client._queue = asyncio.Queue() - task = asyncio.ensure_future(client._consume()) - - for i in range(client_mod._MAX_CONSECUTIVE_DROPS): - client._queue.put_nowait(JobLogDto(Message=str(i))) - await client._queue.join() - sends_while_trying = sends - - client._queue.put_nowait(JobLogDto(Message="muted")) - await client._queue.join() - assert sends == sends_while_trying # dropped while muted, no send attempted - - client._queue.put_nowait(client_mod._STOP) - await task - - -def test_send_log_noop_before_start(): - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - client.send_log(JobLogDto(Message="x")) - - -async def test_send_log_swallows_closed_loop_error(): - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - - class _DeadLoop: - def call_soon_threadsafe(self, *args): - raise RuntimeError("loop closed") - - client._loop = _DeadLoop() # type: ignore[assignment] - client._queue = asyncio.Queue() - client.send_log(JobLogDto(Message="x")) - - -def test_close_noop_before_start(): - client = IpcJobApiClient("pipe", "job", logging.getLogger("test")) - client.close() diff --git a/tests/test_jobapi_contract.py b/tests/test_jobapi_contract.py deleted file mode 100644 index e822ecd..0000000 --- a/tests/test_jobapi_contract.py +++ /dev/null @@ -1,41 +0,0 @@ -import pytest - -from uipath.runtime.jobapi.contract import ( - IIpcLogSink, - JobLogDto, - LogLevel, -) - - -def test_endpoint_name_matches_dotnet_router_key(): - # uipath-ipc routes by the contract class __name__, which must equal the - # .NET interface name the Python executors register (IIpcLogSink, the log - # channel that .NET's IJobInvocationApi extends). - assert IIpcLogSink.__name__ == "IIpcLogSink" - - -def test_contract_only_declares_send_log(): - # Scope is logs only; the result stays on output.json. - assert hasattr(IIpcLogSink, "SendLog") - assert not hasattr(IIpcLogSink, "SetResult") - - -def test_log_level_values_match_microsoft_extensions(): - assert [ - LogLevel.TRACE, - LogLevel.DEBUG, - LogLevel.INFORMATION, - LogLevel.WARNING, - LogLevel.ERROR, - LogLevel.CRITICAL, - LogLevel.NONE, - ] == [0, 1, 2, 3, 4, 5, 6] - - -def test_job_log_dto_wire_shape(): - # The field names are the wire keys, so this pins them against the .NET DTO. - to_wire = pytest.importorskip("uipath_ipc").to_wire - assert to_wire(JobLogDto(Message="hi", LogLevel=3)) == { - "Message": "hi", - "LogLevel": 3, - } diff --git a/tests/test_jobapi_log_handler.py b/tests/test_jobapi_log_handler.py deleted file mode 100644 index 383ebf2..0000000 --- a/tests/test_jobapi_log_handler.py +++ /dev/null @@ -1,57 +0,0 @@ -import logging - -from uipath.runtime.jobapi.contract import JobLogDto, LogLevel -from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, _to_log_level - - -def test_to_log_level_mapping(): - assert _to_log_level(logging.CRITICAL) == LogLevel.CRITICAL - assert _to_log_level(logging.ERROR) == LogLevel.ERROR - assert _to_log_level(logging.WARNING) == LogLevel.WARNING - assert _to_log_level(logging.INFO) == LogLevel.INFORMATION - assert _to_log_level(logging.DEBUG) == LogLevel.DEBUG - # Below DEBUG collapses to TRACE; above CRITICAL clamps to CRITICAL. - assert _to_log_level(1) == LogLevel.TRACE - assert _to_log_level(logging.CRITICAL + 10) == LogLevel.CRITICAL - - -class _RecordingClient: - def __init__(self): - self.logs: list[JobLogDto] = [] - - def send_log(self, dto: JobLogDto) -> None: - self.logs.append(dto) - - -def test_emit_forwards_formatted_message_and_level(): - client = _RecordingClient() - handler = IpcSendLogHandler(client) # type: ignore[arg-type] - handler.setFormatter(logging.Formatter("%(message)s")) - - record = logging.LogRecord( - name="n", - level=logging.WARNING, - pathname="p", - lineno=1, - msg="hello %s", - args=("world",), - exc_info=None, - ) - handler.emit(record) - - assert len(client.logs) == 1 - assert client.logs[0].Message == "hello world" - assert client.logs[0].LogLevel == LogLevel.WARNING - - -def test_emit_swallows_client_errors(): - class _Boom: - def send_log(self, dto: JobLogDto) -> None: - raise RuntimeError("pipe down") - - handler = IpcSendLogHandler(_Boom()) # type: ignore[arg-type] - handler.setFormatter(logging.Formatter("%(message)s")) - record = logging.LogRecord("n", logging.INFO, "p", 1, "x", None, None) - - # handleError writes to sys.stderr but must not raise. - handler.emit(record) diff --git a/tests/test_jobapi_pooled.py b/tests/test_jobapi_pooled.py deleted file mode 100644 index 6785989..0000000 --- a/tests/test_jobapi_pooled.py +++ /dev/null @@ -1,66 +0,0 @@ -import logging -from typing import Any - -import pytest - -from uipath.runtime.jobapi.contract import JobLogDto, LogLevel -from uipath.runtime.jobapi.log_handler import PooledIpcSendLogHandler -from uipath.runtime.jobapi.pooled import get_pooled_log_sink, set_pooled_log_sink - - -@pytest.fixture(autouse=True) -def _clear_sink(): - # The sink is process-global; make sure a test never leaks it to the next. - set_pooled_log_sink(None) - yield - set_pooled_log_sink(None) - - -def test_sink_registry_set_and_clear(): - assert get_pooled_log_sink() is None - - def sink(job_id: str, log: JobLogDto) -> None: - pass - - set_pooled_log_sink(sink) - assert get_pooled_log_sink() is sink - - set_pooled_log_sink(None) - assert get_pooled_log_sink() is None - - -def test_pooled_handler_forwards_tagged_with_job_id(): - calls: list[tuple[str, Any]] = [] - handler = PooledIpcSendLogHandler( - "job-key-1", lambda jid, log: calls.append((jid, log)) - ) - handler.setFormatter(logging.Formatter("%(message)s")) - - record = logging.LogRecord( - name="n", - level=logging.WARNING, - pathname="p", - lineno=1, - msg="hello %s", - args=("world",), - exc_info=None, - ) - handler.emit(record) - - assert len(calls) == 1 - job_id, log = calls[0] - assert job_id == "job-key-1" - assert log.Message == "hello world" - assert log.LogLevel == LogLevel.WARNING - - -def test_pooled_handler_swallows_sink_errors(): - def boom(job_id: str, log: JobLogDto) -> None: - raise RuntimeError("pipe down") - - handler = PooledIpcSendLogHandler("job-key-1", boom) - handler.setFormatter(logging.Formatter("%(message)s")) - record = logging.LogRecord("n", logging.INFO, "p", 1, "x", None, None) - - # handleError writes to stderr but must not raise. - handler.emit(record) diff --git a/tests/test_output_sinks.py b/tests/test_output_sinks.py new file mode 100644 index 0000000..94f83cf --- /dev/null +++ b/tests/test_output_sinks.py @@ -0,0 +1,184 @@ +"""The in-memory output sinks and how the context routes to them. + +By default the runtime writes logs to execution.log and the result to output.json. When a host +installs a log handler / result sink (uipath-python does, to forward over IPC), the runtime routes +to them instead: logs to the handler, and the result — output arguments spilled to a file — to the +sink. Suspended keeps output.json. +""" + +import json +import logging +from pathlib import Path +from typing import Any + +import pytest + +from uipath.runtime.context import UiPathRuntimeContext +from uipath.runtime.output_sinks import ( + get_log_handler, + get_result_sink, + set_log_handler, + set_result_sink, +) +from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus + + +@pytest.fixture(autouse=True) +def _clear_sinks(): + set_log_handler(None) + set_result_sink(None) + yield + set_log_handler(None) + set_result_sink(None) + + +class _DummyInterceptor: + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.log_handler = kwargs.get("log_handler") + + def setup(self) -> None: + pass + + def teardown(self) -> None: + pass + + +@pytest.fixture(autouse=True) +def _patch_interceptor(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "uipath.runtime.context.UiPathRuntimeLogsInterceptor", _DummyInterceptor + ) + + +def _ctx(tmp_path: Path) -> UiPathRuntimeContext: + return UiPathRuntimeContext( + job_id="job-key", runtime_dir=str(tmp_path / "rt"), result_file="output.json" + ) + + +def test_registry_set_and_clear() -> None: + assert get_log_handler() is None + assert get_result_sink() is None + handler = logging.NullHandler() + set_log_handler(handler) + assert get_log_handler() is handler + set_log_handler(None) + assert get_log_handler() is None + + +def test_installed_log_handler_is_used(tmp_path: Path) -> None: + handler = logging.NullHandler() + set_log_handler(handler) + ctx = _ctx(tmp_path) + with ctx: + pass + assert ctx.logs_interceptor.log_handler is handler + + +def test_no_log_handler_defaults_to_file(tmp_path: Path) -> None: + ctx = _ctx(tmp_path) + with ctx: + pass + assert ctx.logs_interceptor.log_handler is None + + +def test_result_sink_receives_result_and_args_file(tmp_path: Path) -> None: + got: list[tuple[Any, str]] = [] + + def _sink(result: Any, path: str) -> None: + got.append((result, path)) + + set_result_sink(_sink) + ctx = _ctx(tmp_path) + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} + ) + + assert len(got) == 1 + result, path = got[0] + assert result.status == UiPathRuntimeStatus.SUCCESSFUL + # The output arguments were spilled to the file the sink was handed. + assert json.loads(Path(path).read_text()) == {"foo": "bar"} + # output.json is still written as a fallback. + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + assert content["output"] == {"foo": "bar"} + + +def test_result_sink_not_called_for_suspended(tmp_path: Path) -> None: + calls: list[int] = [] + + def _sink(result: Any, path: str) -> None: + calls.append(1) + + set_result_sink(_sink) + ctx = _ctx(tmp_path) + with ctx: + ctx.result = UiPathRuntimeResult(status=UiPathRuntimeStatus.SUSPENDED) + + assert calls == [] + + +def test_no_result_sink_leaves_result_on_file(tmp_path: Path) -> None: + ctx = _ctx(tmp_path) + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"a": 1} + ) + + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + assert content["output"] == {"a": 1} + + +def test_result_sink_exception_does_not_clobber_output_json(tmp_path: Path) -> None: + """A sink failure is a side-channel failure: it must not fault the job or rewrite output.json. + + The sink (e.g. IPC delivery) runs after the authoritative output.json is written. A raise here + must be swallowed, NOT caught by __exit__'s catch-all — which would overwrite the good result with + a FAULTED shutdown error. + """ + + def _boom(result: Any, path: str) -> None: + raise RuntimeError("ipc delivery failed") + + set_result_sink(_boom) + ctx = _ctx(tmp_path) + # No exception escapes the context even though the sink raises. + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"ok": 1} + ) + + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + # output.json still holds the SUCCESSFUL result — not clobbered as FAULTED (which drops output + # and adds a RUNTIME_SHUTDOWN_ERROR). + assert content["output"] == {"ok": 1} + assert "error" not in content + + +def test_split_output_arguments_with_sink_does_not_double_write(tmp_path: Path) -> None: + """With split_output_arguments AND a sink, the args file is written once and reused, not twice.""" + got: list[str] = [] + + def _sink(result: Any, path: str) -> None: + got.append(path) + + set_result_sink(_sink) + ctx = UiPathRuntimeContext( + job_id="job-key", + runtime_dir=str(tmp_path / "rt"), + result_file="output.json", + split_output_arguments=True, + ) + with ctx: + ctx.result = UiPathRuntimeResult( + status=UiPathRuntimeStatus.SUCCESSFUL, output={"foo": "bar"} + ) + + # The sink still gets the spilled args file, whose content is the output arguments. + assert len(got) == 1 + assert json.loads(Path(got[0]).read_text()) == {"foo": "bar"} + # output.json carries the pointer (split), not the inline output. + content = json.loads(Path(ctx.resolved_result_file_path).read_text()) + assert content["outputArgumentsFilePath"] == got[0] + assert "output" not in content diff --git a/uv.lock b/uv.lock index 6576d81..7106406 100644 --- a/uv.lock +++ b/uv.lock @@ -7,7 +7,6 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P2D" [options.exclude-newer-package] -uipath-ipc = false uipath-core = false [[package]] @@ -1152,15 +1151,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/37/47a4e12bc7bdcfab4bd4e8e1f2a5c083bd59fca42fd431026a3f14c642a3/uipath_core-0.5.31-py3-none-any.whl", hash = "sha256:3dff5ecb236bf46af683c34d79bb2cf458a942ec041e1cbf7b4825ae246ff00c", size = 55040, upload-time = "2026-07-15T06:56:02.446Z" }, ] -[[package]] -name = "uipath-ipc" -version = "2.5.2" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/2c/6b/53d9725d6abd1dab447300a7f999332f530678e3fabd38fc31171a5a9a6f/uipath_ipc-2.5.2.tar.gz", hash = "sha256:d69c3d7c1ad1a25ef7f9f1d78c505f5d2ed7daa7227bb202404fa3d47e0ba12e", size = 86360, upload-time = "2026-07-24T09:44:35.159Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/29/eb/6def505a0d351119da27b342c11eb0767a31a7f100022d35e349c5194eb3/uipath_ipc-2.5.2-py3-none-any.whl", hash = "sha256:617f25f35377d87956165a875b0966a3cd69ae6724fcd533c93a7a6a9460fc4f", size = 50345, upload-time = "2026-07-24T09:44:33.7Z" }, -] - [[package]] name = "uipath-runtime" version = "0.13.5" @@ -1171,11 +1161,6 @@ dependencies = [ { name = "vadersentiment" }, ] -[package.optional-dependencies] -ipc = [ - { name = "uipath-ipc" }, -] - [package.dev-dependencies] dev = [ { name = "bandit" }, @@ -1189,17 +1174,14 @@ dev = [ { name = "pytest-trio" }, { name = "ruff" }, { name = "rust-just" }, - { name = "uipath-ipc" }, ] [package.metadata] requires-dist = [ { name = "chardet", specifier = ">=5.2.0,<8.0" }, { name = "uipath-core", specifier = ">=0.5.31,<0.6.0" }, - { name = "uipath-ipc", marker = "extra == 'ipc'", specifier = ">=2.5.1,<2.6.0" }, { name = "vadersentiment", specifier = ">=3.3.2,<4.0" }, ] -provides-extras = ["ipc"] [package.metadata.requires-dev] dev = [ @@ -1214,7 +1196,6 @@ dev = [ { name = "pytest-trio", specifier = ">=0.8.0" }, { name = "ruff", specifier = ">=0.9.4" }, { name = "rust-just", specifier = ">=1.39.0" }, - { name = "uipath-ipc", specifier = ">=2.5.1,<2.6.0" }, ] [[package]]