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

Replace the uipath-ipc-coupled jobapi client with transport-agnostic in-memory
sinks (output_sinks.py): the host (uipath-python) installs a log handler and a
result sink, and uipath-runtime keeps no IPC dependency. Drops the [ipc] extra
and the jobapi package.

Review fixes:
- context.__exit__ isolates the result sink so a delivery failure can't clobber
  the already-persisted output.json as a FAULTED shutdown error
- skip the duplicate .args write when split_output_arguments and a sink are both on
- cover that a host-provided (unowned) log handler is not closed on teardown

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: eduard-dumitru

pyproject.toml

@@ -20,10 +20,6 @@ maintainers = [
   { name = "Cristian Pufu", email = "cristian.pufu@uipath.com" },
 ]
 
-[project.optional-dependencies]
-# Job-API IPC log channel; required when UIPATH_JOB_API_IPC_ENDPOINT is set.
-ipc = ["uipath-ipc>=2.5.1, <2.6.0"]
-
 [project.urls]
 Homepage = "https://uipath.com"
 Repository = "https://github.com/UiPath/uipath-runtime-python"
@@ -46,7 +42,6 @@ dev = [
   "pytest-cov>=4.1.0",
   "pytest-mock>=3.11.1",
   "pre-commit>=4.1.0",
-  "uipath-ipc>=2.5.1, <2.6.0",
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -128,7 +123,6 @@ exclude-newer = "2 days"
 
 [tool.uv.exclude-newer-package]
 uipath-core = false
-uipath-ipc = false
 
 [[tool.uv.index]]
 name = "testpypi"

src/uipath/runtime/context.py

@@ -18,10 +18,8 @@
     UiPathErrorContract,
     UiPathRuntimeError,
 )
-from uipath.runtime.jobapi.client import IpcJobApiClient
-from uipath.runtime.jobapi.log_handler import IpcSendLogHandler, PooledIpcSendLogHandler
-from uipath.runtime.jobapi.pooled import get_pooled_log_sink
 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__)
@@ -123,38 +121,8 @@ class UiPathRuntimeContext(BaseModel):
     keep_state_file: bool = Field(
         False, description="Prevents deletion of state file before running."
     )
-    ipc_endpoint: str | None = Field(
-        None,
-        description=(
-            "uipath-ipc endpoint (UIPATH_JOB_API_IPC_ENDPOINT) for streaming logs to "
-            "the handler in place of execution.log. The result stays on output.json."
-        ),
-    )
-    ipc_job_id: str | None = Field(
-        None,
-        description="Handler job id (UIPATH_JOB_ID) used to route the IPC calls.",
-    )
-    ipc_client: Any = Field(default=None, exclude=True, repr=False)
-
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
 
-    @property
-    def ipc_active(self) -> bool:
-        """Whether logs flow over a per-job IPC pipe (non-pooled) instead of to files."""
-        return bool(self.ipc_endpoint and self.ipc_job_id)
-
-    @property
-    def pooled_ipc_active(self) -> bool:
-        """Whether logs stream over the pooled server's callback rather than a per-job pipe.
-
-        True when there is a job id but no endpoint, and the pooled server registered its sink.
-        """
-        return (
-            bool(self.ipc_job_id)
-            and not self.ipc_endpoint
-            and get_pooled_log_sink() is not None
-        )
-
     def _apply_execution_source(self) -> None:
         """Derive execution_source from the command, if not already set.
 
@@ -270,22 +238,9 @@ def __enter__(self):
         Returns:
             The runtime context instance
         """
-        # Intercept all stdout/stderr/logs
-        # Write to file (runtime), stdout (debug) or log handler (if provided)
-        log_handler: logging.Handler | None = None
-        if self.ipc_active:
-            assert self.ipc_endpoint is not None and self.ipc_job_id is not None
-            self.ipc_client = IpcJobApiClient(
-                self.ipc_endpoint, self.ipc_job_id, logger
-            )
-            self.ipc_client.start()
-            log_handler = IpcSendLogHandler(self.ipc_client)
-            log_handler.setFormatter(logging.Formatter("%(message)s"))
-        elif self.pooled_ipc_active:
-            sink = get_pooled_log_sink()
-            assert self.ipc_job_id is not None and sink is not None
-            log_handler = PooledIpcSendLogHandler(self.ipc_job_id, sink)
-            log_handler.setFormatter(logging.Formatter("%(message)s"))
+        # 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,
@@ -359,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
 
@@ -402,8 +372,21 @@ def __exit__(self, exc_type, exc_val, exc_tb):
             # Restore original logging
             if hasattr(self, "logs_interceptor"):
                 self.logs_interceptor.teardown()
-            if self.ipc_client is not None:
-                self.ipc_client.close()
+
+    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:
@@ -468,8 +451,6 @@ def with_defaults(
         base.tenant_id = os.environ.get("UIPATH_TENANT_ID")
         base.process_key = os.environ.get("UIPATH_PROCESS_UUID")
         base.folder_key = os.environ.get("UIPATH_FOLDER_KEY")
-        base.ipc_endpoint = os.environ.get("UIPATH_JOB_API_IPC_ENDPOINT")
-        base.ipc_job_id = os.environ.get("UIPATH_JOB_ID")
 
         # Override with kwargs
         for k, v in kwargs.items():