feat: optionally split output arguments into their own file

eduard-dumitru · claude · eduard-dumitru · commit 2c879630db8a · 2026-08-11T17:52:19.000+02:00
Adds runtime.splitOutputArguments to uipath.json. When set, the output arguments
are written next to the result file and the envelope carries an absolute
outputArgumentsFilePath pointer instead of the inline output value, so a consumer
can stream that file rather than materializing it.

Opt-in and default-off: with the knob unset the emitted output is byte-identical
to before. A failing write degrades to inline and never changes job status, and
status/error/resume/resumeTriggers always stay inline.

Co-Authored-By: Claude Opus 5 &lt;noreply@anthropic.com&gt;
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.13.0"
+version = "0.13.1"
 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
@@ -23,6 +23,8 @@
 
 logger = logging.getLogger(__name__)
 
+OUTPUT_ARGUMENTS_FILE_NAME = "output.args.json"
+
 _EXECUTION_SOURCE_BY_COMMAND: dict[str, str] = {
     "run": "runtime",
     "debug": "playground",
@@ -87,6 +89,14 @@ class UiPathRuntimeContext(BaseModel):
             "If not specified, path is constructed from runtime_dir and result_file."
         ),
     )
+    split_output_arguments: bool = Field(
+        False,
+        description=(
+            "Write the output arguments to their own file alongside the result file, "
+            "and carry an 'outputArgumentsFilePath' pointer in the result file "
+            "instead of the inline 'output' value."
+        ),
+    )
     state_file: str = Field("state.db", description="Filename for the state database")
     state_file_path: str | None = Field(
         None,
@@ -275,6 +285,27 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
             content = self.result.to_dict()
 
+            # Captured before the pop below, so output_file still gets the real args
+            output_payload = content.get("output", {})
+
+            # Only an optimization, so a failing write degrades to the inline value
+            # rather than changing the outcome of an otherwise successful run
+            if self.split_output_arguments:
+                try:
+                    output_arguments_path = self.resolved_output_arguments_file_path
+                    os.makedirs(os.path.dirname(output_arguments_path), exist_ok=True)
+                    with open(output_arguments_path, "w") as f:
+                        json.dump(output_payload, f, default=str)
+                except Exception as e:
+                    logger.warning(
+                        "Failed to write the output arguments file, keeping the output arguments inline: %s",
+                        e,
+                        exc_info=True,
+                    )
+                else:
+                    content.pop("output", None)
+                    content["outputArgumentsFilePath"] = output_arguments_path
+
             # Always write output file at runtime, except for inner runtimes
             # Inner runtimes have execution_id
             if self.job_id:
@@ -283,7 +314,6 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
             # Write the execution output to file if requested
             if self.output_file:
-                output_payload = content.get("output", {})
                 with open(self.output_file, "w") as f:
                     json.dump(output_payload, f, default=str)
 
@@ -343,6 +373,19 @@ def resolved_result_file_path(self) -> str:
             return os.path.join(self.runtime_dir, self.result_file)
         return os.path.join("__uipath", "output.json")
 
+    @cached_property
+    def resolved_output_arguments_file_path(self) -> str:
+        """Get the full path to the output arguments file.
+
+        Derived, not configured: the name is fixed and the directory is the result
+        file's, so the host cannot put the two files in different places and the
+        knob has exactly one encoding.
+        """
+        return os.path.join(
+            os.path.dirname(os.path.abspath(self.resolved_result_file_path)),
+            OUTPUT_ARGUMENTS_FILE_NAME,
+        )
+
     @cached_property
     def resolved_state_file_path(self) -> str:
         """Get the full path to the state file."""
@@ -406,6 +449,7 @@ def from_config(
         mapping = {
             "dir": "runtime_dir",
             "outputFile": "result_file",  # we need this to maintain back-compat with serverless runtime
+            "splitOutputArguments": "split_output_arguments",
             "stateFile": "state_file",
             "logsFile": "logs_file",
         }
diff --git a/tests/test_context.py b/tests/test_context.py
@@ -1,9 +1,12 @@
 import json
+import logging
+import os
 from pathlib import Path
 from typing import Any
 
 import pytest
 from uipath.core.errors import ErrorCategory, UiPathFaultedTriggerError
+from uipath.core.triggers import UiPathResumeTrigger
 
 from uipath.runtime.context import UiPathRuntimeContext
 from uipath.runtime.errors import (
@@ -427,3 +430,240 @@ def test_from_config_accepts_maestro_flow_voice_mode(tmp_path: Path) -> None:
     ctx = UiPathRuntimeContext.from_config(str(config_path))
 
     assert ctx.voice_mode == "maestro_flow"
+
+
+def test_from_config_maps_split_output_arguments(tmp_path: Path) -> None:
+    """runtime.splitOutputArguments should map onto the knob."""
+    cfg = {"runtime": {"splitOutputArguments": True}}
+    config_path = tmp_path / "uipath.json"
+    config_path.write_text(json.dumps(cfg))
+
+    ctx = UiPathRuntimeContext.from_config(config_path=str(config_path))
+
+    assert ctx.split_output_arguments is True
+
+
+def test_split_output_arguments_defaults_off_when_config_key_absent(
+    tmp_path: Path,
+) -> None:
+    """The split stays off when the config omits the key."""
+    cfg = {"runtime": {"outputFile": "my_output.json"}}
+    config_path = tmp_path / "uipath.json"
+    config_path.write_text(json.dumps(cfg))
+
+    ctx = UiPathRuntimeContext.from_config(config_path=str(config_path))
+
+    assert ctx.split_output_arguments is False
+
+
+def test_output_arguments_file_is_a_sibling_of_the_result_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The arguments file lands next to the result file, never in the process CWD.
+
+    The host names the directory once, through runtime_dir, and both files follow
+    it. The filename is not configurable, so the knob has exactly one encoding and
+    the two files cannot be pointed at different directories.
+    """
+    cwd = tmp_path / "cwd"
+    cwd.mkdir()
+    monkeypatch.chdir(cwd)
+    runtime_dir = tmp_path / "runtime"
+    ctx = UiPathRuntimeContext(
+        job_id="job-sibling",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        split_output_arguments=True,
+    )
+
+    arguments_path = Path(ctx.resolved_output_arguments_file_path)
+    assert arguments_path.parent == Path(ctx.resolved_result_file_path).parent
+    assert arguments_path.parent == runtime_dir
+    assert arguments_path.name == "output.args.json"
+    assert arguments_path.is_absolute()
+    assert cwd not in arguments_path.parents
+
+
+def test_result_file_keeps_output_inline_when_split_disabled(
+    tmp_path: Path,
+) -> None:
+    """Without the knob, the result file is byte-identical to the legacy envelope."""
+    runtime_dir = tmp_path / "runtime"
+    ctx = UiPathRuntimeContext(
+        job_id="job-inline",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+    )
+
+    with ctx:
+        ctx.result = UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"foo": "bar"},
+        )
+
+    result_path = Path(ctx.resolved_result_file_path)
+    # The envelope is written in text mode, so json's newline reaches disk as os.linesep
+    expected = json.dumps(
+        {"output": {"foo": "bar"}, "status": "successful"}, indent=2
+    ).replace("\n", os.linesep)
+    assert result_path.read_bytes() == expected.encode()
+
+    content = json.loads(result_path.read_bytes())
+    assert "outputArgumentsFilePath" not in content
+    assert not Path(ctx.resolved_output_arguments_file_path).exists()
+
+
+def test_output_arguments_written_to_separate_file(tmp_path: Path) -> None:
+    """With the knob, the arguments move out and the envelope carries the path."""
+    runtime_dir = tmp_path / "nested" / "runtime"
+    ctx = UiPathRuntimeContext(
+        job_id="job-split",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        split_output_arguments=True,
+    )
+
+    with ctx:
+        ctx.result = UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"foo": "bar"},
+        )
+
+    arguments_path = Path(ctx.resolved_output_arguments_file_path)
+    # Parent directory is created on demand
+    assert json.loads(arguments_path.read_text()) == {"foo": "bar"}
+
+    content = json.loads(Path(ctx.resolved_result_file_path).read_text())
+    assert "output" not in content
+    assert content["status"] == UiPathRuntimeStatus.SUCCESSFUL.value
+    assert content["outputArgumentsFilePath"] == str(arguments_path)
+    assert Path(content["outputArgumentsFilePath"]).is_absolute()
+
+
+def test_output_file_receives_bare_arguments_when_split_enabled(
+    tmp_path: Path,
+) -> None:
+    """--output-file keeps receiving the bare arguments when both are set."""
+    runtime_dir = tmp_path / "runtime"
+    output_path = tmp_path / "output.json"
+    ctx = UiPathRuntimeContext(
+        job_id="job-both",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        output_file=str(output_path),
+        split_output_arguments=True,
+    )
+
+    with ctx:
+        ctx.result = UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+            output={"foo": "bar"},
+        )
+
+    assert json.loads(output_path.read_text()) == {"foo": "bar"}
+    arguments_path = Path(ctx.resolved_output_arguments_file_path)
+    assert json.loads(arguments_path.read_text()) == {"foo": "bar"}
+
+
+def test_faulted_run_keeps_status_and_error_inline_when_split_enabled(
+    tmp_path: Path,
+) -> None:
+    """status and error stay in the envelope when the arguments are split out."""
+    runtime_dir = tmp_path / "runtime"
+    ctx = UiPathRuntimeContext(
+        job_id="job-faulted-split",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        split_output_arguments=True,
+    )
+
+    with pytest.raises(RuntimeError, match="Stream blew up"):
+        with ctx:
+            raise RuntimeError("Stream blew up")
+
+    content = json.loads(Path(ctx.resolved_result_file_path).read_text())
+    assert content["status"] == UiPathRuntimeStatus.FAULTED.value
+    assert content["error"]["code"] == "ERROR_RuntimeError"
+    assert "Stream blew up" in content["error"]["detail"]
+    assert "output" not in content
+
+    # The pointer must never advertise a file that was not actually written
+    arguments_path = Path(content["outputArgumentsFilePath"])
+    assert arguments_path.exists()
+    assert json.loads(arguments_path.read_text()) == {}
+
+
+def test_resume_triggers_stay_inline_when_split_enabled(tmp_path: Path) -> None:
+    """resume and resumeTriggers must never be moved out of the envelope.
+
+    They are what makes a suspended job resumable, so a split that swept them
+    into the arguments file would strand the job.
+    """
+    runtime_dir = tmp_path / "runtime"
+    ctx = UiPathRuntimeContext(
+        job_id="job-suspended-split",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        split_output_arguments=True,
+    )
+
+    trigger = UiPathResumeTrigger(item_key="k")
+    with ctx:
+        ctx.result = UiPathRuntimeResult(
+            status=UiPathRuntimeStatus.SUSPENDED,
+            output={"foo": "bar"},
+            trigger=trigger,
+            triggers=[trigger],
+        )
+
+    content = json.loads(Path(ctx.resolved_result_file_path).read_text())
+    assert content["status"] == UiPathRuntimeStatus.SUSPENDED.value
+    assert content["resume"]["itemKey"] == "k"
+    assert len(content["resumeTriggers"]) == 1
+    assert content["resumeTriggers"][0]["itemKey"] == "k"
+    # Only the output moved out
+    assert "output" not in content
+    arguments_path = Path(content["outputArgumentsFilePath"])
+    assert json.loads(arguments_path.read_text()) == {"foo": "bar"}
+
+
+def test_failed_arguments_write_degrades_to_inline_output(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failing arguments write keeps the output inline and the run successful.
+
+    Contract: splitting the arguments out is only an optimization, so a failed
+    write must never turn an otherwise successful run into a faulted one.
+    """
+    runtime_dir = tmp_path / "runtime"
+    runtime_dir.mkdir()
+    # A directory cannot be opened for writing, so the split write fails
+    (runtime_dir / "output.args.json").mkdir()
+    ctx = UiPathRuntimeContext(
+        job_id="job-degraded-write",
+        runtime_dir=str(runtime_dir),
+        result_file="result.json",
+        split_output_arguments=True,
+    )
+
+    with caplog.at_level(logging.WARNING, logger="uipath.runtime.context"):
+        with ctx:
+            ctx.result = UiPathRuntimeResult(
+                status=UiPathRuntimeStatus.SUCCESSFUL,
+                output={"foo": "bar"},
+            )
+
+    content = json.loads(Path(ctx.resolved_result_file_path).read_text())
+    assert content["status"] == UiPathRuntimeStatus.SUCCESSFUL.value
+    assert content["output"] == {"foo": "bar"}
+    assert "outputArgumentsFilePath" not in content
+    assert "error" not in content
+
+    record = next(
+        r
+        for r in caplog.records
+        if "Failed to write the output arguments file" in r.message
+    )
+    # The cause is interpolated and the traceback preserved, not just the prefix
+    assert "%s" not in record.message
+    assert record.exc_info is not None
diff --git a/uv.lock b/uv.lock
