diff --git a/pyproject.toml b/pyproject.toml index c624e4c..ebe7206 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" diff --git a/src/uipath/runtime/context.py b/src/uipath/runtime/context.py index 575db03..ad79866 100644 --- a/src/uipath/runtime/context.py +++ b/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 @@ 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") def _apply_execution_source(self) -> None: @@ -238,13 +238,16 @@ def __enter__(self): 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 @@ 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 @@ -355,6 +373,21 @@ 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 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.""" 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_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_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 a0dd246..7106406 100644 --- a/uv.lock +++ b/uv.lock @@ -1153,7 +1153,7 @@ wheels = [ [[package]] name = "uipath-runtime" -version = "0.13.4" +version = "0.13.5" source = { editable = "." } dependencies = [ { name = "chardet" },