feat: stream pooled job logs to the handler over uipath-ipc

packages/uipath/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.14.10"
+version = "2.14.11"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath/src/uipath/_cli/_job_api.py

@@ -0,0 +1,336 @@
+"""The uipath-ipc job-invocation contract + the glue that routes a job's logs and result over it.
+
+This lives in uipath-python (which owns the ``uipath-ipc`` dependency and the IPC connection).
+uipath-runtime stays transport-agnostic: it just calls the in-memory sinks installed here via
+``uipath.runtime.output_sinks``. ``install_runtime_sinks`` points those sinks at an
+``IJobInvocationApi`` callback (the handler, reached over the same pipe); ``clear_runtime_sinks``
+restores the default file behaviour.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import sys
+import threading
+from abc import ABC, abstractmethod
+from collections.abc import AsyncIterator
+from concurrent.futures import Future
+from dataclasses import dataclass
+from enum import IntEnum
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_SET_RESULT_TIMEOUT_S = 30.0
+
+
+class LogLevel(IntEnum):
+    """Mirror of Microsoft.Extensions.Logging.LogLevel (the wire values)."""
+
+    TRACE = 0
+    DEBUG = 1
+    INFORMATION = 2
+    WARNING = 3
+    ERROR = 4
+    CRITICAL = 5
+    NONE = 6
+
+
+class ExecutorJobStatus(IntEnum):
+    """Mirror of the handler's ExecutorJobStatus (the wire values)."""
+
+    RUNNING = 1
+    FAULTED = 2
+    SUCCESSFUL = 3
+    STOPPED = 4
+    SUSPENDED = 5
+
+
+@dataclass
+class JobLogDto:
+    """A single log entry (PascalCase to match the wire; .NET has no [JsonProperty])."""
+
+    Message: str = ""
+    LogLevel: int = LogLevel.INFORMATION.value
+
+
+@dataclass
+class JobExecutorError:
+    """The result's error (PascalCase; the .NET JobExecutorError has no [JsonProperty])."""
+
+    Code: str | None = None
+    Title: str | None = None
+    Detail: str | None = None
+    Category: str | None = None
+    Status: int | None = None
+
+
+@dataclass
+class JobResultDto:
+    """The final job result (camelCase to match the .NET [JsonProperty] wire keys).
+
+    ``outputArguments`` (the unbounded customer output) is never sent inline — the runtime spills it
+    to a file and this carries only ``outputArgumentsFilePath``. The rest is the bounded envelope.
+    """
+
+    id: str = ""
+    status: int = ExecutorJobStatus.SUCCESSFUL.value
+    outputArguments: Any = None
+    outputArgumentsFilePath: str | None = None
+    info: str | None = None
+    error: JobExecutorError | None = None
+
+
+class IIpcLogSink(ABC):
+    """The log channel; the class name is the CoreIpc endpoint key for the logs-only base."""
+
+    @abstractmethod
+    async def SendLog(self, jobId: str, log: JobLogDto) -> None:
+        """Forward one log entry (one-way)."""
+
+
+class IJobInvocationApi(IIpcLogSink):
+    """The full contract the handler hosts: logs (inherited) plus the final result."""
+
+    @abstractmethod
+    async def SetResult(self, jobId: str, result: JobResultDto) -> bool:
+        """Submit the final job result (request-response: the handler acks)."""
+
+
+def _to_log_level(levelno: int) -> int:
+    if levelno >= logging.CRITICAL:
+        return LogLevel.CRITICAL
+    if levelno >= logging.ERROR:
+        return LogLevel.ERROR
+    if levelno >= logging.WARNING:
+        return LogLevel.WARNING
+    if levelno >= logging.INFO:
+        return LogLevel.INFORMATION
+    if levelno >= logging.DEBUG:
+        return LogLevel.DEBUG
+    return LogLevel.TRACE
+
+
+_EXECUTOR_STATUS: dict[str, int] = {
+    "successful": ExecutorJobStatus.SUCCESSFUL.value,
+    "faulted": ExecutorJobStatus.FAULTED.value,
+    "suspended": ExecutorJobStatus.SUSPENDED.value,
+}
+
+
+def _to_result_dto(
+    job_id: str, result: Any, output_arguments_file_path: str
+) -> JobResultDto:
+    """Map a uipath-runtime result to the wire DTO (envelope inline, output arguments as a pointer)."""
+    error = None
+    if result is not None and getattr(result, "error", None) is not None:
+        category = result.error.category
+        error = JobExecutorError(
+            Code=result.error.code,
+            Title=result.error.title,
+            Detail=result.error.detail,
+            Category=getattr(category, "value", category),
+            Status=result.error.status,
+        )
+    raw_status = getattr(result, "status", None)
+    status_key = str(getattr(raw_status, "value", raw_status) or "successful").lower()
+    return JobResultDto(
+        id=job_id,
+        status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value),
+        outputArgumentsFilePath=output_arguments_file_path,
+        error=error,
+    )
+
+
+def _drain(future: "Future[object]") -> None:
+    try:
+        future.exception()
+    except BaseException:
+        pass
+
+
+class _IpcLogHandler(logging.Handler):
+    """Forwards each record to the handler's SendLog over the callback (one-way, non-blocking)."""
+
+    def __init__(
+        self, job_id: str, callback: Any, loop: asyncio.AbstractEventLoop
+    ) -> None:
+        super().__init__()
+        self._job_id = job_id
+        self._callback = callback
+        self._loop = loop
+
+    def emit(self, record: logging.LogRecord) -> None:
+        try:
+            message = self.format(record)
+            dto = JobLogDto(Message=message, LogLevel=_to_log_level(record.levelno))
+            future = asyncio.run_coroutine_threadsafe(
+                self._callback.SendLog(self._job_id, dto), self._loop
+            )
+            future.add_done_callback(_drain)
+        except Exception:
+            self.handleError(record)
+
+
+def install_runtime_sinks(
+    job_id: str, callback: Any, loop: asyncio.AbstractEventLoop
+) -> None:
+    """Point uipath-runtime's log handler + result sink at ``callback`` (an IJobInvocationApi proxy).
+
+    Logs are forwarded one-way; the result blocks for the handler's ack. Both schedule onto ``loop``
+    (the connection's loop), which MUST be running on a different thread than the one the runtime
+    invokes the sinks on — otherwise the blocking ack would wait on a loop that can never run it
+    (deadlock). Pooled: the job runs on a worker thread while ``loop`` is the free server loop;
+    non-pooled: ``loop`` is a dedicated thread (see ``connect_handler_ipc``). No-op if uipath-runtime
+    lacks the sinks.
+    """
+    try:
+        from uipath.runtime.output_sinks import (  # type: ignore[import-untyped]
+            set_log_handler,
+            set_result_sink,
+        )
+    except ImportError:
+        return
+
+    handler = _IpcLogHandler(job_id, callback, loop)
+    handler.setFormatter(logging.Formatter("%(message)s"))
+
+    def _result_sink(result: Any, output_arguments_file_path: str) -> None:
+        dto = _to_result_dto(job_id, result, output_arguments_file_path)
+        try:
+            future = asyncio.run_coroutine_threadsafe(
+                callback.SetResult(job_id, dto), loop
+            )
+            future.result(timeout=_SET_RESULT_TIMEOUT_S)
+        except Exception:
+            # Best-effort delivery: don't fault the job (output.json still holds the result), but a
+            # dropped IPC result must be observable, not silent.
+            logger.exception("Failed to deliver job result over IPC (SetResult)")
+
+    set_log_handler(handler)
+    set_result_sink(_result_sink)
+
+
+def clear_runtime_sinks() -> None:
+    """Restore uipath-runtime's default file behaviour (execution.log / output.json)."""
+    try:
+        from uipath.runtime.output_sinks import set_log_handler, set_result_sink
+    except ImportError:
+        return
+    set_log_handler(None)
+    set_result_sink(None)
+
+
+def _new_ipc_event_loop() -> asyncio.AbstractEventLoop:
+    """A fresh event loop for the handler IPC connection to run on its own thread.
+
+    On Windows the named-pipe client needs the Proactor loop (``create_pipe_connection``);
+    ``new_event_loop()`` already yields it there, but build it explicitly so a non-default event
+    loop policy can't hand back a Selector loop that cannot dial a pipe.
+    """
+    if sys.platform == "win32":
+        return asyncio.ProactorEventLoop()
+    return asyncio.new_event_loop()
+
+
+class _HandlerIpcConnection:
+    """A handler IPC client bound to its OWN event loop on its OWN thread.
+
+    ``uipath run`` enters the runtime context inline under ``asyncio.run(...)``, so the runtime
+    invokes the result sink synchronously on the job's loop thread. The sink blocks for the
+    handler's ack (``future.result()``); if that ack were scheduled onto the job's own loop it could
+    never run — deadlock. Running the connection on a dedicated loop/thread keeps the two apart: the
+    sink blocks the job thread while this loop delivers the call (mirroring the pooled path, where
+    the job already runs on a worker thread while the server loop is free).
+    """
+
+    def __init__(
+        self,
+        client: Any,
+        loop: asyncio.AbstractEventLoop,
+        thread: threading.Thread,
+    ) -> None:
+        self._client = client
+        self._loop = loop
+        self._thread = thread
+
+    def _shutdown(self) -> None:
+        """Close the client on its loop, then stop the loop and join its thread (best-effort)."""
+        try:
+            asyncio.run_coroutine_threadsafe(self._client.aclose(), self._loop).result(
+                timeout=_SET_RESULT_TIMEOUT_S
+            )
+        except Exception:
+            pass
+        self._loop.call_soon_threadsafe(self._loop.stop)
+        self._thread.join(timeout=_SET_RESULT_TIMEOUT_S)
+        self._loop.close()
+
+
+def connect_handler_ipc(pipe: str, job_id: str) -> _HandlerIpcConnection:
+    """Dial the handler's per-job pipe on a dedicated loop/thread and install the runtime sinks.
+
+    Non-pooled counterpart of the pooled callback: ``uipath run`` runs in its own process, so it
+    dials the handler's per-job server itself (the handler passed the pipe as ``--handler-ipc-pipe``)
+    rather than reaching back over an existing connection. The connection gets its OWN loop on its
+    OWN thread so the result-sink ack — which the runtime invokes inline on the job's loop thread —
+    cannot deadlock the loop it waits on (see ``_HandlerIpcConnection``).
+    """
+    try:
+        from uipath_ipc import IpcClient, NamedPipeClientTransport
+    except ImportError as e:
+        raise RuntimeError(
+            "The handler asked for the uipath-ipc job-api channel (--handler-ipc-pipe) but the "
+            "'uipath-ipc' package is not installed. Install it (pip install 'uipath[ipc]')."
+        ) from e
+
+    loop = _new_ipc_event_loop()
+    thread = threading.Thread(
+        target=loop.run_forever, name="uipath-handler-ipc", daemon=True
+    )
+    thread.start()
+
+    async def _build() -> Any:
+        client = IpcClient(transport=NamedPipeClientTransport(pipe))
+        proxy = client.get_proxy(IJobInvocationApi)  # type: ignore[type-abstract]
+        # Point the sinks at THIS dedicated loop, not the caller's — that is the fix: the sink's
+        # blocking ack now waits on a loop running on another thread, which is free to run it.
+        install_runtime_sinks(job_id, proxy, loop)
+        return client
+
+    try:
+        client = asyncio.run_coroutine_threadsafe(_build(), loop).result(
+            timeout=_SET_RESULT_TIMEOUT_S
+        )
+    except BaseException:
+        # Building the client failed; don't leak the loop/thread we just started.
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(timeout=_SET_RESULT_TIMEOUT_S)
+        loop.close()
+        raise
+    return _HandlerIpcConnection(client, loop, thread)
+
+
+async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None:
+    """Clear the runtime sinks and tear down the connection's loop/thread (best-effort)."""
+    clear_runtime_sinks()
+    # Tear down off the caller's loop so joining the connection thread doesn't block it.
+    await asyncio.to_thread(conn._shutdown)
+
+
+@contextlib.asynccontextmanager
+async def handler_ipc_connection(pipe: str | None, job_id: str) -> AsyncIterator[Any]:
+    """Connect the handler IPC leg for the run (if ``pipe`` is set) and ALWAYS disconnect on exit.
+
+    Yields the connection (or None when no pipe was given). Guarantees the sinks are cleared and the
+    connection torn down even if the job raises — a bare connect/disconnect pair would skip that on
+    the exception path.
+    """
+    conn = connect_handler_ipc(pipe, job_id) if pipe else None
+    try:
+        yield conn
+    finally:
+        if conn is not None:
+            await disconnect_handler_ipc(conn)

packages/uipath/src/uipath/_cli/_server_core.py

@@ -3,7 +3,7 @@
 import asyncio
 import os
 import shlex
-from typing import Any
+from typing import Any, Callable
 
 from .cli_debug import debug
 from .cli_eval import eval
@@ -50,8 +50,15 @@ async def _run_command_isolated(
     args: list[str],
     env_vars: dict[str, str],
     working_dir: str | None,
+    on_run_start: Callable[[], None] | None = None,
+    on_run_end: Callable[[], None] | None = None,
 ) -> dict[str, Any]:
-    """Run one command with per-job env/cwd isolation (the shared job core)."""
+    """Run one command with per-job env/cwd isolation (the shared job core).
+
+    ``on_run_start`` / ``on_run_end`` run INSIDE the serialization lock, immediately around the job,
+    so per-job process-global state (e.g. the IPC output sinks) is visible ONLY while this job runs —
+    a concurrently-dispatched job cannot observe or clear another job's state.
+    """
     if _state.lock is None or _state.baseline_env is None:
         raise RuntimeError("Server state not initialized")
 
@@ -79,9 +86,15 @@ async def _run_command_isolated(
                         "ClientError": True,
                     }
 
-            result_value = await asyncio.to_thread(
-                cmd.main, args, standalone_mode=False
-            )
+            if on_run_start is not None:
+                on_run_start()
+            try:
+                result_value = await asyncio.to_thread(
+                    cmd.main, args, standalone_mode=False
+                )
+            finally:
+                if on_run_end is not None:
+                    on_run_end()
             return {
                 "ExitCode": 0,
                 "Error": None,