feat: stream pooled job logs to the handler over uipath-ipc - #1883
Open
eduard-dumitru wants to merge 2 commits into
Open
feat: stream pooled job logs to the handler over uipath-ipc#1883eduard-dumitru wants to merge 2 commits into
eduard-dumitru wants to merge 2 commits into
Conversation
The pooled `uipath server` now grabs the handler's IIpcLogSink callback during Register (via the injected Message) and points uipath-runtime's process-global log sink at it, so a pooled job's logs stream back over the existing pipe. Best-effort and version-guarded: an older uipath-runtime without the pooled sink API is a no-op and jobs keep the file path. Register(self, message) gains the reach-back handle; the sink forwards each SendLog onto the server loop from the job's worker thread. Bump 2.14.10 -> 2.14.11. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
The pooled log forwarding path is documented as best-effort but currently can surface exceptions during callback acquisition/forwarding, and the new test includes a timing-based sleep that can be flaky.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR adds the “pooled” half of IPC log streaming by wiring a pooled job’s process-global runtime log sink to an IIpcLogSink callback provided by the handler over the existing uipath-ipc connection. The Register handshake is extended to receive an injected Message (reach-back handle) so the server can obtain the callback and forward logs from worker threads onto the server event loop.
Changes:
- Extend the IPC contract so
Register(self, message)receives the injecteduipath_ipc.Messageand uses it to wire a pooled log sink callback. - Add server-side forwarding logic (
run_coroutine_threadsafe+ drain) and lifecycle cleanup to unset the process-global sink when the IPC server exits. - Add tests that lock in the callback seam behavior and bump package version to
2.14.11.
File summaries
| File | Description |
|---|---|
| packages/uipath/src/uipath/_cli/cli_server_ipc.py | Add Message-aware Register, wire pooled log sink to handler callback, and unset sink on server shutdown. |
| packages/uipath/tests/cli/test_server_ipc.py | Update contract test for new Register signature and add pooled log sink seam tests. |
| packages/uipath/pyproject.toml | Bump uipath version from 2.14.10 to 2.14.11. |
Review details
- Files reviewed: 3/3 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+151
to
+158
| sink_proxy = client.get_callback(IIpcLogSink) | ||
| loop = asyncio.get_running_loop() | ||
|
|
||
| def _forward(job_id: str, log: object) -> None: | ||
| # Runs on the job's worker thread: hand the one-way SendLog to the server loop and return at | ||
| # once. A dropped log (pipe down) must never surface into the job, so failures are swallowed. | ||
| future = asyncio.run_coroutine_threadsafe(sink_proxy.SendLog(job_id, log), loop) | ||
| future.add_done_callback(_drain) |
Comment on lines
+389
to
+393
| sink = captured["sink"] | ||
| log = {"Message": "hello", "LogLevel": 2} | ||
| sink("job-key-42", log) | ||
| await asyncio.sleep(0.05) | ||
| assert sent == [("job-key-42", log)] |
uipath-python owns the uipath-ipc contract and connection (_job_api.py) and points uipath-runtime's in-memory sinks at it. Pooled: `uipath server` grabs the handler's IJobInvocationApi callback at Register and installs the sinks per job. Non-pooled: `uipath run --handler-ipc-pipe` dials the handler's per-job server. The result envelope travels inline; output arguments go off-heap via a file pointer. Review fixes: - run the non-pooled handler connection on its own loop/thread so the result sink's blocking ack can't deadlock the job's own event loop - install/clear the process-global sinks inside the job core's lock so concurrently-dispatched RunJobs can't corrupt each other's routing - log (not silently swallow) an IPC result-delivery failure - always disconnect the handler IPC on the run's exit path (async context manager) - register the default runtime factory in start_ipc_server - pin the DTO wire key sets and the log-level mapping in tests Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
The pooled
uipath servernow streams each job's execution logs back to the Robot handler over the existing CoreIPC pipe, as a client callback — the pooled half of the log-streaming endeavor. Logs only; the result stays onoutput.json.How
Register(self, message)gains the injectedMessagereach-back handle. On register, the server grabsmessage.client.get_callback(IIpcLogSink)and points uipath-runtime's process-global log sink at it; eachSendLogis forwarded onto the server loop from the job's worker thread. Jobs run in-process (_run_command_isolated→asyncio.to_thread, serialized), so a process-global sink genuinely reaches the callback.Best-effort and version-guarded: if
uipath.runtime.jobapiisn't present (older runtime), the wiring is a no-op and jobs keep the file path.Registerstill returns True.Version
2.14.10 → 2.14.11.Related PRs — one endeavor (this depends on the runtime PR)
Merge order: uipath-runtime → this PR → hdens.
Testing
Server-IPC tests pass, including 3 new tests for the callback seam (
TestPooledLogSink); ruff/format/mypy clean. (test_run_job_successfails only in a bare local venv with no runtime factory — pre-existing, not this change.)Jira
Tracking ticket: ROBO-5981.
🤖 Generated with Claude Code