feat(runtime): in-memory output sinks for host-driven log/result delivery

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

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
 from uipath.core.errors import UiPathFaultedTriggerError
 from uipath.core.tracing import UiPathTraceManager
 
@@ -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,9 +121,13 @@ class UiPathRuntimeContext(BaseModel):
     keep_state_file: bool = Field(
         False, description="Prevents deletion of state file before running."
     )
-
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
 
+    # Snapshot of the result sink taken at __enter__ and reused at __exit__, so the same sink that was
+    # installed when the context started is the one that receives the result — even if the registry is
+    # changed in between or __exit__ runs on a different event loop.
+    _result_sink: ResultSink | None = PrivateAttr(default=None)
+
     def _apply_execution_source(self) -> None:
         """Derive execution_source from the command, if not already set.
 
@@ -238,13 +243,17 @@ def __enter__(self):
         Returns:
             The runtime context instance
         """
-        # Intercept all stdout/stderr/logs
-        # Write to file (runtime), stdout (debug) or log handler (if provided)
+        # Snapshot both caller-installed sinks now, at context start. The log handler is consumed here;
+        # the result sink is stashed for __exit__ so the two are read at the same moment (same context).
+        log_handler = get_log_handler()
+        self._result_sink = get_result_sink()
+
         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 +320,19 @@ def __exit__(self, exc_type, exc_val, exc_tb):
                 with open(self.output_file, "w") as f:
                     json.dump(output_payload, f, default=str)
 
+            # Best-effort side channel: a sink failure must NOT reach the catch-all below, which would
+            # rewrite the already-good output.json as FAULTED. Reuse the __enter__ snapshot, not a fresh
+            # read, so the sink is the one that was installed when the context started.
+            result_sink = self._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:
+                    logger.exception("Failed to deliver result to sink")
+
             # Don't suppress exceptions
             return False
 
@@ -355,6 +377,16 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             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 sink."""
+        args_path = self.resolved_output_arguments_file_path
+        # Avoid re-spilling if split_output_arguments already wrote this file.
+        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 @@
+"""Per-execution slots for a log handler and a result sink a caller can install.
+
+Unset, the runtime writes its usual files; set, it routes a job's logs and result to them instead.
+Values are contextvars, isolated across concurrent jobs in one process.
+"""
+
+from __future__ import annotations
+
+import logging
+from contextvars import ContextVar
+from typing import Any, Callable
+
+# (result, output_arguments_file_path) -> None
+ResultSink = Callable[[Any, str], None]
+
+_log_handler: ContextVar[logging.Handler | None] = ContextVar(
+    "uipath_log_handler", default=None
+)
+_result_sink: ContextVar[ResultSink | None] = ContextVar(
+    "uipath_result_sink", default=None
+)
+
+
+def set_log_handler(handler: logging.Handler | None) -> None:
+    """Install the log handler (``None`` clears)."""
+    _log_handler.set(handler)
+
+
+def get_log_handler() -> logging.Handler | None:
+    """The installed log handler, or None."""
+    return _log_handler.get()
+
+
+def set_result_sink(sink: ResultSink | None) -> None:
+    """Install the result sink (``None`` clears)."""
+    _result_sink.set(sink)
+
+
+def get_result_sink() -> ResultSink | None:
+    """The installed result sink, or None."""
+    return _result_sink.get()

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