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

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"

src/uipath/runtime/context.py

@@ -19,6 +19,7 @@
     UiPathRuntimeError,
 )
 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__)
@@ -120,7 +121,6 @@
     keep_state_file: bool = Field(
         False, description="Prevents deletion of state file before running."
     )
-
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
 
     def _apply_execution_source(self) -> None:
@@ -238,13 +238,16 @@
         Returns:
             The runtime context instance
         """
-        # Intercept all stdout/stderr/logs
-        # Write to file (runtime), stdout (debug) or log handler (if provided)
+        # 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,
             dir=self.runtime_dir,
             file=self.logs_file,
             job_id=self.job_id,
+            log_handler=log_handler,
         )
         self.logs_interceptor.setup()
 
@@ -311,6 +314,21 @@
                 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
 
@@ -355,6 +373,21 @@
             if hasattr(self, "logs_interceptor"):
                 self.logs_interceptor.teardown()
 
+    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:
         """Get the full path to the result file."""

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

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."""
 

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