feat(runtime): in-memory output sinks for host-driven log/result delivery - #167
feat(runtime): in-memory output sinks for host-driven log/result delivery#167eduard-dumitru wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
🟡 Changes recommended
Some newly added documentation/comments contradict the PR’s “logs-only” behavior, and the IPC client startup can block indefinitely due to an unbounded readiness wait.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds a new uipath.runtime.jobapi surface to stream execution logs to the handler over uipath-ipc (non-pooled per-job client and pooled server callback path), and wires UiPathRuntimeContext to use it when the relevant env vars/sink are present—while keeping job results on output.json.
Changes:
- Introduce
jobapipackage: IPC contract (IIpcLogSink+ DTO), non-pooled IPC client, pooled sink registry, and logging handlers. - Update
UiPathRuntimeContextto inject an IPC-backed log handler (suppressingexecution.log) when IPC is active / pooled sink is registered. - Add
[ipc]optional dependency and add tests covering contract shape, handlers, pooled registry, client round-trip, and context branching.
File summaries
| File | Description |
|---|---|
| uv.lock | Locks uipath-ipc and bumps package version to 0.13.5; adds ipc extra metadata. |
| pyproject.toml | Bumps version to 0.13.5; adds [project.optional-dependencies].ipc and dev dep for uipath-ipc. |
| src/uipath/runtime/context.py | Adds IPC env wiring and injects IPC/pooled log handlers into the logs interceptor. |
| src/uipath/runtime/jobapi/init.py | Exposes the new jobapi public surface via __all__. |
| src/uipath/runtime/jobapi/client.py | Implements per-job IPC client with private asyncio loop/thread, FIFO queue, retry/mute, and flush-on-close. |
| src/uipath/runtime/jobapi/contract.py | Defines the IPC contract (IIpcLogSink) and DTO/enums mirroring the .NET wire shape. |
| src/uipath/runtime/jobapi/log_handler.py | Adds logging.Handler implementations to forward records via IPC client or pooled sink. |
| src/uipath/runtime/jobapi/pooled.py | Adds process-global pooled log sink registry for pooled server integration. |
| tests/test_context_ipc.py | Verifies context selects IPC vs pooled vs file logging paths and keeps result on output.json. |
| tests/test_jobapi_client.py | Validates optional-dep failure mode and does an IPC named-pipe round-trip test. |
| tests/test_jobapi_contract.py | Pins wire contract naming and DTO serialization shape. |
| tests/test_jobapi_log_handler.py | Tests handler formatting/level mapping and error swallowing. |
| tests/test_jobapi_pooled.py | Tests pooled sink registry and pooled handler tagging/error swallowing. |
Review details
Suppressed comments (1)
src/uipath/runtime/context.py:412
- The teardown comment references sending the result over IPC and a
set_resultflush, but this context still writes results tooutput.jsonandIpcJobApiClienthas noset_result. The comment is now inaccurate/outdated.
# Tear down the IPC channel after logging is restored (the result was
# already sent above; set_result flushed the queued logs first).
if self.ipc_client is not None:
self.ipc_client.close()
- Files reviewed: 12/13 changed files
- Comments generated: 3
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| self._thread = threading.Thread( | ||
| target=self._run, name="uipath-jobapi-ipc", daemon=True | ||
| ) | ||
| self._thread.start() | ||
| self._ready.wait() | ||
| if self._start_error is not None: | ||
| raise self._start_error |
| [project.optional-dependencies] | ||
| # The job-API IPC channel (stream logs + report result to the handler). Optional | ||
| # so only serverless installs that use it pull the transport; required whenever | ||
| # UIPATH_JOB_API_IPC_ENDPOINT is set (the runtime fails loudly if it is missing). | ||
| ipc = ["uipath-ipc>=2.5.1, <2.6.0"] |
| ipc_endpoint: str | None = Field( | ||
| None, | ||
| description=( | ||
| "uipath-ipc endpoint (UIPATH_JOB_API_IPC_ENDPOINT) to stream logs and " | ||
| "report the result back to the handler, replacing the log/result files." | ||
| ), | ||
| ) |
225ffa4 to
1848a1b
Compare
Add a jobapi package that forwards a job's execution logs to the Robot handler over
uipath-ipc instead of execution.log, for both non-pooled and pooled executors. Logs
only; the result stays on output.json (kept off-heap by split_output_arguments).
- contract.py: IIpcLogSink (SendLog) + JobLogDto{Message, LogLevel}, mirroring .NET.
- client.py: IpcJobApiClient, the non-pooled per-job client (private asyncio thread,
Proactor on Windows, FIFO queue with retry/mute).
- pooled.py + log_handler.py: a process-global PooledLogSink and PooledIpcSendLogHandler
for the pooled server, which owns the CoreIPC callback one layer up.
- context.py: use the IPC handler when UIPATH_JOB_API_IPC_ENDPOINT + UIPATH_JOB_ID are set
(non-pooled), or the pooled sink when only UIPATH_JOB_ID is set and a sink is registered;
execution.log is suppressed in both cases.
Optional [ipc] extra (uipath-ipc). Bump 0.13.4 -> 0.13.5.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
88f6cd3 to
383acc0
Compare
eduard-dumitru
changed the title
383acc0 to
1e09897
Compare
|
|
||
| def set_log_handler(handler: "logging.Handler | None") -> None: | ||
| """Install the log handler (``None`` clears).""" | ||
| global _log_handler |
There was a problem hiding this comment.
the global keyword here is a symptom of a process-global mutable registry. The runtime reads the log handler during __enter__, but reads the result sink later during __exit__. If another execution replaces or clears the registry in between, logs and results could be routed to different hosts, or a result could be delivered to the wrong sink.
we should instead pass a small RuntimeOutputSinks dependency into UiPathRuntimeContext or provide a scoped install/reset context manager.
There was a problem hiding this comment.
I employed contextvars to differentiate between potential concurrent jobs. I think that falls under a scoped install/reset context manager suggestion.
Anything touching existing public surface would need to be debated (like parameters in existing constructors).
| with open(args_path, "w") as f: | ||
| json.dump(output_payload, f, default=str) |
There was a problem hiding this comment.
nit: we can (and should) use the async file apis
note: this would imply adding
__aexit__/__aenter__methods
There was a problem hiding this comment.
It's true I've increased it a bit, but sync-IO in __exit__ predates this endeavor. We should talk about changing the public surface.
1e09897 to
d4747ca
Compare
…very 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>
d4747ca to
f0939d9
Compare
|
What
Add
runtime/output_sinks.py— a tiny process-global registry of two in-memory sinks a host can install so a job's logs and result flow to the host's own transport instead of the files. The runtime stays transport-agnostic (no new dependency); nothing installs the sinks here — that's the host's job.execution.logis suppressed.(result, output_arguments_file_path).output.jsonstays the fallback and the source of truth for Suspended/resume.Related PRs — one endeavor (merge in this order)
Three PRs, one endeavor — stream Python serverless job logs + result to the .NET Robot handler over CoreIPC:
Testing
441 tests green; ruff/format/mypy clean.
Jira
Tracking: ROBO-5980.
🤖 Generated with Claude Code