Repository navigation
feat(cli): stream job logs and result to the handler over uipath-ipc - #1883
Conversation
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.
| 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) |
| sink = captured["sink"] | ||
| log = {"Message": "hello", "LogLevel": 2} | ||
| sink("job-key-42", log) | ||
| await asyncio.sleep(0.05) | ||
| assert sent == [("job-key-42", log)] |
8b4869d to
1dc7057
Compare
eduard-dumitru
changed the title
e9af2ca to
bea284f
Compare
robert-ursu
left a comment
robert-ursu
left a comment
There was a problem hiding this comment.
Reviewed against the PR head, the uipath-runtime 0.13.5 / uipath-ipc 2.5.2 sources, and the hdens#8031 C# side.
Main theme: the new IPC path removes the execution.log fallback but has no way to notice when the channel is broken, so logs can vanish while the job still reports success. The two blocking items are the stale Register callback on the pooled server and the lazy connect on the non-pooled path. Details inline.
| try: | ||
| from ._job_api import IJobInvocationCommonApi | ||
|
|
||
| self._callback = client.get_callback(IJobInvocationCommonApi) # type: ignore[type-abstract] |
There was a problem hiding this comment.
The callback is captured once on the singleton service, but message.client is the per-accept connection. hdens calls Register only from Initialize() and never re-issues it, so after a pipe reconnect (_callback proxies a closed connection) or a Python server restart (_callback is None) every opted-in job for the rest of the pod's life either streams into a dead connection or skips streaming. Since hdens no longer watches execution.log for these jobs, their logs vanish with no signal.
Suggest declaring RunJob(self, request, message: Message) and deriving message.client.get_callback(...) plus the running loop per request, instead of storing connection state on the service.
| callback, loop = self._callback, self._loop | ||
| on_run_start: "Any" = None | ||
| on_run_end: "Any" = None | ||
| if request.StreamOutputOverIpc and callback is not None and loop is not None: |
There was a problem hiding this comment.
When the handler opted the job in (it has already skipped the execution.log watcher) but no callback/loop is available here, the job silently runs with file sinks nobody reads. The mismatch is detectable at this point; it should surface as an error in the RunJob result (or at least a warning) rather than fall through.
| # Drop the sinks so they can't outlive this loop (matters on in-process restart). | ||
| from ._job_api import clear_runtime_sinks | ||
|
|
||
| clear_runtime_sinks() |
There was a problem hiding this comment.
This is a no-op: the sinks are ContextVars set inside the per-request handler task (uipath-ipc dispatches each request via asyncio.create_task), so clearing them from the server task's context can't affect them. Same for on_run_end=clear_runtime_sinks. The "process-global sinks" framing in the comment and PR description is wrong; suggest dropping this and fixing the comment so the next reader doesn't rely on it.
| pass | ||
|
|
||
|
|
||
| class _IpcLogHandler(logging.Handler): |
There was a problem hiding this comment.
No self-exclusion filter. uipath_ipc is imported before the interceptor attaches this handler to every logger, so uipath_ipc.client.connection's own diagnostics (dropping malformed RESPONSE frame, receive loop failed, ...) become JobLogDto entries in the customer's job output, and each one schedules another SendLog on the very transport whose failure it reports. Suggest a filter dropping uipath_ipc.* and this module's records (route them to the original stderr).
| future = asyncio.run_coroutine_threadsafe( | ||
| self._callback.SendLog(self._job_id, dto), self._loop | ||
| ) | ||
| future.add_done_callback(_drain) |
There was a problem hiding this comment.
The SendLog futures are never awaited, so the tail of the log can be dropped. Non-pooled: disconnect_handler_ipc -> aclose() fails any still-queued sends with ConnectionError, which _drain swallows. Pooled: the RunJob response is written right after on_run_end with no wait for acks, so .NET can Unregister the job before the last frames arrive (the sink then logs "Dropped pooled Python job log for unregistered job"). Track the futures on the handler and do a bounded wait before aclose() / before returning from RunJob.
| pass | ||
| self._loop.call_soon_threadsafe(self._loop.stop) | ||
| self._thread.join(timeout=_SET_RESULT_TIMEOUT_S) | ||
| self._loop.close() |
There was a problem hiding this comment.
If join(timeout=30) times out because the IPC thread is wedged (peer half-open, aclose() hanging), loop.close() raises RuntimeError: Cannot close a running event loop. Inside the context manager's finally that replaces the job's real exception, or turns a successful run into exit 1. Same issue on the failure path in connect_handler_ipc (line 266), where the bare raise never executes and the original TimeoutError is masked. Guard close() on thread.is_alive() and make teardown non-raising.
| thread.start() | ||
|
|
||
| async def _build() -> Any: | ||
| client = IpcClient(transport=NamedPipeClientTransport(pipe)) |
There was a problem hiding this comment.
NamedPipeClientTransport connects lazily on the first call, so nothing here verifies the pipe is reachable. But installing the handler makes UiPathRuntimeLogsInterceptor skip the execution.log file handler, so an unreachable pipe (POSIX $TMPDIR mismatch, peer not yet bound, typo) loses 100% of the job's logs while uipath run exits 0 with "Successful execution". Do one eager round-trip at connect and fail loudly, or fall back to the file handler on first failure.
Also: no request_timeout on IpcClient. Against a peer that accepted the socket but stalls, every log line becomes an immortal task parked on the write lock with no cap, for the whole job.
| [project] | ||
| name = "uipath" | ||
| version = "2.14.10" | ||
| version = "2.14.11" |
There was a problem hiding this comment.
2.14.11 is already on PyPI (main is at 2.14.14), and the branch conflicts with main on pyproject.toml and uv.lock. Needs a rebase and a bump to >= 2.14.15. Worth getting right because the hdens VenvIpcCapability version floors will be set from this release: if they end up at 2.14.11, pods with 2.14.11-2.14.14 venvs would get StreamOutputOverIpc / --handler-ipc-pipe, the watcher skipped, and an older uipath that ignores or rejects the option.
| from ._job_api import handler_ipc_connection | ||
|
|
||
| async with ( | ||
| handler_ipc_connection(handler_ipc_pipe, ctx.job_id or ""), |
There was a problem hiding this comment.
ctx.job_id or "" sends an empty string where the .NET contract has Guid jobId. Without UIPATH_JOB_KEY (manual/dev runs) Newtonsoft can't convert "" to Guid, so every SendLog/SetResult errors and, given the issues above, nothing is reported. Skip connecting, or fail fast with a clear message, when ctx.job_id is None.
| debug_port: int, | ||
| keep_state_file: bool, | ||
| simulation: str | None, | ||
| handler_ipc_pipe: str | None, |
There was a problem hiding this comment.
handler_ipc_pipe isn't forwarded to Middlewares.next("run", ...), and the sinks are installed only inside execute(). A run middleware that returns should_continue=False executes the job with the flag silently ignored, while the handler has already skipped the execution.log watcher. Either pass it through the middleware chain or refuse the flag when a run middleware takes over.
777996f to
0aa6a56
Compare
github-actions
Bot
added
test:uipath-langchain
4608cd0 to
de514dd
Compare
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>
uipath-python owns the uipath-ipc contract and connection (_job_api.py) and points uipath-runtime's in-memory sinks at it. The contract is IJobInvocationCommonApi (SendLog + SetResult), the runtime-agnostic base the handler hosts; the JS-only telemetry channel lives on a .NET superset Python doesn't use, so Python keys on the common base directly. Pooled: `uipath server` grabs the handler's IJobInvocationCommonApi 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>
The IPC log/result sinks this CLI installs live in uipath.runtime.output_sinks, which first shipped in uipath-runtime 0.13.5. The old >=0.13.1 floor (and a lockfile pinned to 0.13.1) would resolve a runtime without that module, silently no-opping the feature via the ImportError fallback. Bump the floor and relock to 0.13.5. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ping
The _to_result_dto tests set the fake result's status to bare strings
("faulted"/"successful"), but the runtime hands a UiPathRuntimeStatus
(str, Enum) member, where str(member) is "UiPathRuntimeStatus.FAULTED",
not "faulted". Only the getattr(status, "value", status) extraction
normalizes it — and the bare-string fakes never took that path, so
dropping it would report a faulted/suspended job as SUCCESSFUL while
every test stayed green.
Use real UiPathRuntimeStatus members in the fakes and add the SUSPENDED
case (previously untested), so the mapping is guarded against that
regression.
…Register Nothing exercised the failure guarantees the design forbids violating: a SendLog forward that fails must not escape emit(), a SetResult failure must be logged and not raised (else the job faults and output.json is clobbered), and a get_callback that raises at Register must leave the callback None rather than break registration. Add a test for each. Verified by mutation: making any of the three except blocks re-raise fails the matching test.
The IPC pooled-sink tests fake _run_command_isolated, so the hook-placement guarantees were only asserted against the fake. Drive real on_run_start/on_run_end spies through the real core and pin: on_run_end fires in the finally even when the job raises (SystemExit/Exception), on_run_start runs only after env/cwd are applied, and neither hook runs on the bad-working-dir early return. Verified by mutation: relocating either hook fails these tests.
…gister Register captured the caller's callback on the singleton service, but message.client is the per-accept connection and hdens only calls Register from Initialize(). After a pipe reconnect the stored callback proxied a closed connection, and after a Python server restart it was None — so every opted-in job for the rest of the pod's life either streamed into a dead connection or skipped streaming. With execution.log no longer watched for those jobs, their logs vanished with no signal. RunJob now takes the injected Message and derives the callback plus the running loop from that request's own connection; Register is a readiness probe and the service holds no connection state. Also: - Fail loud when a job is opted in but no callback is available, rather than running it with file sinks nobody reads. - Drop clear_runtime_sinks() from start_ipc_server's finally: uipath-ipc dispatches each request in its own task, so the sinks are per-request contextvars and clearing them from the server task was a no-op. - Remove a type: ignore that the uipath-runtime 0.13.5 bump made unused. - Capture the result-delivery log on the module logger directly; through caplog it was flaky, since other tests in the suite mutate propagation.
The runtime interceptor attaches the job log handler to the root logger, so
uipath_ipc's own diagnostics ("dropping malformed frame", "receive loop
failed", ...) became JobLogDto entries in the customer's job output — and
each one scheduled another SendLog on the very transport whose failure it
was reporting.
Route records from uipath_ipc.* and this module to the original stderr
instead of the IPC channel, so they stay visible in the pod without riding
the channel they describe.
Also correct the Register docstring: it still claimed to grab the caller's
callback, which moved to RunJob.
SendLog futures were fire-and-forget, so the tail of a job's log could be lost: non-pooled, aclose() failed whatever was still queued; pooled, the RunJob response was written straight after the job, letting the peer unregister it before the last frames arrived. Track pending sends on the handler and drain them, bounded, before teardown. The two paths need different drains: non-pooled blocks inside _shutdown (it already runs on a worker thread, and the sink loop has its own), while pooled must await in RunJob, because on_run_end runs on the very loop the sends need in order to complete.
Two teardown-path defects on the job-api channel. SetResult's return value was discarded, but the handler returns false when it cannot route the job (unknown job key, or the executor threw). A rejected result was therefore treated as delivered and silently lost; it is now reported. The report reaches the original stderr rather than the channel that just rejected it. loop.close() also ran unconditionally after a timeout-bounded join. A wedged IPC thread leaves the loop running, so close() raised: inside the context manager's finally that masked the job's real exception (or turned a success into exit 1), and on the connect-failure path it pre-empted the bare raise, masking the original error. Teardown now lives in one non-raising helper used by both paths, which only closes the loop once its thread is gone.
NamedPipeClientTransport dials on first use, so an unreachable pipe (a TMPDIR mismatch, a peer not yet bound, a typo) was not noticed at connect. By then the interceptor has already suppressed execution.log for the job, so every log line went nowhere while `uipath run` still exited 0 reporting a successful execution. Dial the transport eagerly and raise with the pipe name if it cannot be reached. Losing the job is better than running it with its entire log silently discarded. Also give IpcClient a request timeout: against a peer that accepts the connection and then stalls, each log line otherwise became an uncancellable task parked on the write lock for the life of the job. connect_handler_ipc is therefore no longer lazy - it now requires a reachable pipe, and its test stands up a real server instead of relying on the old behaviour.
The peer types the job id as a Guid and rejects anything it cannot parse as one. Both lanes accepted whatever they were handed: the CLI turned a missing UIPATH_JOB_KEY into "", and the pooled server passed an unset JobKey straight through. Since requesting IPC suppresses the file sinks, every SendLog and SetResult then failed on the wire and the run produced no logs anywhere while still exiting 0. Validate the shape in both places and fail fast instead.
A "run" middleware that returns should_continue=False runs the job itself and skips the block that opens the pipe, so the caller -- which stopped watching the log file because it was promised a pipe -- got no logs, no result, and "Successful execution.". Refuse the flag in that case instead. Both checks now test the pipe for presence rather than truthiness, so an empty name cannot slip past them either.
Nineteen comments down to seven one-liners. What is left is the blocking/awaiting asymmetry between the two lanes, why teardown must not raise, why the connect is eager, where the contextvar sinks have to be installed, why the Message placeholder survives import, and why the callback is never cached. The rest restated the code or repeated an error message.
…orks The dispatcher builds its plan from the contract, not the implementation, so declaring `message` only on PythonRuntimeService left it uninjected: every pooled job that opted into streaming was rejected with "no IPC callback is available" before it ran. Every pooled test passed a message by hand, which hid it -- the new test drives a real pipe with a client that hosts the callback.
SendLog failures landed in a future whose exception was retrieved and thrown away, so a handler that went away mid-job produced an empty log stream and exit 0 -- the failure this channel exists to prevent. Report the first one to the real stderr, then stay quiet so a dead pipe cannot spam.
Refuse an all-zeros job id: Guid's default parses as a UUID, so it passed the shape check and then routed nowhere, the peer dropping every log on a debug line nobody reads. The test meant to cover it used a 35-character string, so it had only ever exercised "malformed". Hide --handler-ipc-pipe: the job executor sets it, a person never does, yet it was listed in every user's --help and in the generated CLI docs. Drop the "no IPC callback available" guard down to what is knowable. The dispatcher always injects a Message carrying the connection and get_callback merely wraps it, so the guard could not fire over IPC while advertising a wire-level problem it was incapable of seeing. Stop claiming the serialization lock bounds per-job global state: cancelling the RunJob task releases it while the job is still running on a thread cancellation cannot reach. On the tests, four that did not pin what they claimed: - The sink helper swapped the runtime's ContextVars for a dict, recording THAT something was installed but not WHERE. Installing from a foreign thread -- the one thing the code forbids -- kept all 42 green; it now fails two. - The non-pooled test proved the pipe was dialled, not that the job ran inside it. Hoisting the connection out of the run, or passing None for the job id, both passed; both now fail. - No test asserted a log line crossing a real pipe, only the result half. - The mapped error checked Code and Category alone, so swapping Title and Detail passed -- a stack trace would surface as the error's title. Plus coverage for aflush_pending and for a cancelled send, whose future raises rather than returns, and a note that the suspended mapping is defensive: the runtime delivers SUCCESSFUL and FAULTED alone and the peer resolves suspend from output.json.
Refuse to stream when the runtime has no output sinks. install_runtime_sinks swallowed the ImportError and returned None, and neither lane noticed: the connection came back live with no log handler and the job reported ExitCode 0, while the peer had already stopped watching execution.log on our promise. The whole job's output vanished and everything reported success. Teardown still tolerates an older runtime; only install refuses. Match the peer's frame cap. uipath-ipc caps an inbound frame at 2 MB, but every .NET server on this contract already accepts 30 -- PythonJobApiServer for the per-job pipe, IpcServerDefaults for the JS lane. Python was the only endpoint still at the default. Give the timeouts room. 30s for a call and 5s to drain in-flight sends were picked with nothing behind them. They are ceilings, not delays -- a healthy peer acks in under a millisecond -- so they only decide how long to hang on when the peer is slow. Nothing throttles emit and production p99 is 7,500 log lines a job (worst seen 58,694), so a backed-up peer can have thousands in flight when the job ends, and five seconds truncates exactly the tail worth having. 120s for all three. Three tests did not pin what they claimed: - An earlier commit deleted the line `class TestMiddleware:`, reparenting five unrelated middleware tests under TestHandlerIpcPipe, so `-k TestMiddleware` selected nothing. - The non-pooled deadlock regression passed WITH the deadlock, in 30.9s instead of 0.19s: every assertion sat after `await disconnect_handler_ipc`, so the sink burned its timeout, gave up, and the queued ack completed during that await. Assert before any await, and bound the call. - The pooled real-pipe test patched install_runtime_sinks away, so the callback it obtained was never used -- pointing get_callback at another contract kept the suite, mypy and ruff clean while every pooled send went to an endpoint the peer does not host. Install for real and assert a line arrives. Also: drive the public context manager rather than connect/disconnect, since cli_run only ever uses that; pin that the sinks are live when the runtime context snapshots them; and clear what the batching revert left behind.
44d0fb2 to
f880ae0
Compare
🚨 Heads up:
|
|
What
uipath-python owns the uipath-ipc contract and connection (
_job_api.py) and points uipath-runtime's in-memory sinks (from #167) at the handler, so a Python serverless job streams its logs and result back over CoreIPC instead of the files. Covers both transport roles. Version2.14.10 → 2.14.11.Two roles, one contract —
IJobInvocationCommonApi(SendLog+SetResult)uipath server) —Register(message)grabs the handler's callback (message.client.get_callback(IJobInvocationCommonApi)) on the existing pooled pipe and installs the runtime sinks for the job. Gated per-job by aStreamOutputOverIpcflag on the run request; jobs run in-process (serialized), so the process-global sinks genuinely reach the callback.uipath run --handler-ipc-pipe) — dials the handler's per-job server (get_proxy) and installs the sinks on a dedicated loop/thread, so the result's blocking ack can't deadlock the job's own event loop.SendLogcarries logs;SetResultcarries a bounded envelope + a pointer to the output-arguments file the runtime spilled (the payload never crosses IPC). Best-effort and version-guarded: a no-op if the runtime lacksoutput_sinksor uipath-ipc is absent, so an older runtime keeps the files.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
CLI/IPC tests green, incl. the non-pooled result-ack (deadlock) regression + DTO wire-key tests; ruff/format/mypy clean.
Jira
Tracking: ROBO-5981.
🤖 Generated with Claude Code