Skip to content

Commit 5d69e08

Browse files
feat(cli): speak the handler's IPythonJobApi contract, naming the run on every message
The handler now hosts a Python-specific job api instead of sharing the JS one, and it routes pooled callbacks by job key and resume version. Every log and result DTO carries JobKey and ResumeVersion inline; the log DTO also carries the record's UTC timestamp so the handler can stamp the line with when it was emitted rather than when it arrived. Wire keys are PascalCase throughout, matching the peer's property names. Gated off in production until the handler raises its version floor to this release.
1 parent 3e18950 commit 5d69e08

6 files changed

Lines changed: 166 additions & 120 deletions

File tree

‎packages/uipath/pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[project]
22
name = "uipath"
3-
version = "2.14.23"
3+
version = "2.14.24"
44
description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
55
readme = { file = "README.md", content-type = "text/markdown" }
66
requires-python = ">=3.11"

‎packages/uipath/src/uipath/_cli/_job_api.py‎

Lines changed: 71 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
"""The job-invocation IPC contract and the glue that routes a job's logs and result over it."""
1+
"""The Python job-api IPC contract and the glue that routes a job's logs and result over it."""
22

33
from __future__ import annotations
44

@@ -13,6 +13,7 @@
1313
from concurrent.futures import Future
1414
from concurrent.futures import wait as _futures_wait
1515
from dataclasses import dataclass
16+
from datetime import datetime, timezone
1617
from enum import IntEnum
1718
from typing import Any, TypeGuard
1819

@@ -55,11 +56,15 @@ class ExecutorJobStatus(IntEnum):
5556

5657

5758
@dataclass
58-
class JobLogDto:
59+
class PythonJobLogDto:
5960
"""A log entry; field names are the wire keys (do not rename)."""
6061

62+
JobKey: str = ""
63+
ResumeVersion: int | None = None
6164
Message: str = ""
6265
LogLevel: int = LogLevel.INFORMATION.value
66+
# ISO-8601 UTC, when the record was emitted; the peer may stamp the log with it.
67+
Timestamp: str | None = None
6368

6469

6570
@dataclass
@@ -74,26 +79,31 @@ class JobExecutorError:
7479

7580

7681
@dataclass
77-
class JobResultDto:
82+
class PythonJobResultDto:
7883
"""The final result; field names are the wire keys (do not rename)."""
7984

80-
id: str = ""
81-
status: int = ExecutorJobStatus.SUCCESSFUL.value
82-
outputArguments: Any = None
83-
outputArgumentsFilePath: str | None = None
84-
info: str | None = None
85-
error: JobExecutorError | None = None
85+
JobKey: str = ""
86+
ResumeVersion: int | None = None
87+
Status: int = ExecutorJobStatus.SUCCESSFUL.value
88+
OutputArguments: Any = None
89+
OutputArgumentsFilePath: str | None = None
90+
Info: str | None = None
91+
Error: JobExecutorError | None = None
8692

8793

88-
class IJobInvocationCommonApi(ABC):
89-
"""The job-invocation contract: logs + the final result. The class name is the endpoint key."""
94+
class IPythonJobApi(ABC):
95+
"""The Python job-api contract: logs + the final result. The class name is the endpoint key.
96+
97+
Every message names the run it belongs to (job key + resume version), so the peer can route a
98+
pooled callback to the right job and drop a straggler from a previous resume.
99+
"""
90100

91101
@abstractmethod
92-
async def SendLog(self, jobId: str, log: JobLogDto) -> None:
102+
async def SendLog(self, log: PythonJobLogDto) -> None:
93103
"""Forward one log entry."""
94104

95105
@abstractmethod
96-
async def SetResult(self, jobId: str, result: JobResultDto) -> bool:
106+
async def SetResult(self, result: PythonJobResultDto) -> bool:
97107
"""Submit the final result."""
98108

99109

@@ -121,8 +131,11 @@ def _to_log_level(levelno: int) -> int:
121131

122132

123133
def _to_result_dto(
124-
job_id: str, result: Any, output_arguments_file_path: str
125-
) -> JobResultDto:
134+
job_key: str,
135+
resume_version: int | None,
136+
result: Any,
137+
output_arguments_file_path: str,
138+
) -> PythonJobResultDto:
126139
error = None
127140
if result is not None and getattr(result, "error", None) is not None:
128141
category = result.error.category
@@ -135,11 +148,12 @@ def _to_result_dto(
135148
)
136149
raw_status = getattr(result, "status", None)
137150
status_key = str(getattr(raw_status, "value", raw_status) or "successful").lower()
138-
return JobResultDto(
139-
id=job_id,
140-
status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value),
141-
outputArgumentsFilePath=output_arguments_file_path,
142-
error=error,
151+
return PythonJobResultDto(
152+
JobKey=job_key,
153+
ResumeVersion=resume_version,
154+
Status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value),
155+
OutputArgumentsFilePath=output_arguments_file_path,
156+
Error=error,
143157
)
144158

145159

@@ -162,10 +176,15 @@ class _IpcLogHandler(logging.Handler):
162176
"""Forwards each log record to the callback."""
163177

164178
def __init__(
165-
self, job_id: str, callback: Any, loop: asyncio.AbstractEventLoop
179+
self,
180+
job_key: str,
181+
callback: Any,
182+
loop: asyncio.AbstractEventLoop,
183+
resume_version: int | None = None,
166184
) -> None:
167185
super().__init__()
168-
self._job_id = job_id
186+
self._job_key = job_key
187+
self._resume_version = resume_version
169188
self._callback = callback
170189
self._loop = loop
171190
self._pending: set[Future[object]] = set()
@@ -177,11 +196,17 @@ def emit(self, record: logging.LogRecord) -> None:
177196
_to_original_stderr(self, record)
178197
return
179198
try:
180-
dto = JobLogDto(
181-
Message=self.format(record), LogLevel=_to_log_level(record.levelno)
199+
dto = PythonJobLogDto(
200+
JobKey=self._job_key,
201+
ResumeVersion=self._resume_version,
202+
Message=self.format(record),
203+
LogLevel=_to_log_level(record.levelno),
204+
Timestamp=datetime.fromtimestamp(
205+
record.created, tz=timezone.utc
206+
).isoformat(),
182207
)
183208
future = asyncio.run_coroutine_threadsafe(
184-
self._callback.SendLog(self._job_id, dto), self._loop
209+
self._callback.SendLog(dto), self._loop
185210
)
186211
with self._pending_lock:
187212
self._pending.add(future)
@@ -226,7 +251,10 @@ async def aflush_pending(self, timeout: float = _LOG_FLUSH_TIMEOUT_S) -> None:
226251

227252

228253
def install_runtime_sinks(
229-
job_id: str, callback: Any, loop: asyncio.AbstractEventLoop
254+
job_key: str,
255+
callback: Any,
256+
loop: asyncio.AbstractEventLoop,
257+
resume_version: int | None = None,
230258
) -> "_IpcLogHandler | None":
231259
"""Install the log + result sinks, forwarding to ``callback`` on ``loop``.
232260
@@ -246,15 +274,15 @@ def install_runtime_sinks(
246274
"Install uipath-runtime>=0.13.5."
247275
) from e
248276

249-
handler = _IpcLogHandler(job_id, callback, loop)
277+
handler = _IpcLogHandler(job_key, callback, loop, resume_version)
250278
handler.setFormatter(logging.Formatter("%(message)s"))
251279

252280
def _result_sink(result: Any, output_arguments_file_path: str) -> None:
253-
dto = _to_result_dto(job_id, result, output_arguments_file_path)
281+
dto = _to_result_dto(
282+
job_key, resume_version, result, output_arguments_file_path
283+
)
254284
try:
255-
future = asyncio.run_coroutine_threadsafe(
256-
callback.SetResult(job_id, dto), loop
257-
)
285+
future = asyncio.run_coroutine_threadsafe(callback.SetResult(dto), loop)
258286
if future.result(timeout=_SET_RESULT_TIMEOUT_S) is False:
259287
logger.error(
260288
"The handler rejected the job result (SetResult returned false)"
@@ -330,7 +358,7 @@ def _shutdown(self) -> None:
330358

331359

332360
def is_wire_job_id(job_id: str | None) -> TypeGuard[str]:
333-
"""The peer types the job id as a Guid, and routes nothing for one it can't match."""
361+
"""The peer types the job key as a Guid, and routes nothing for one it can't match."""
334362
try:
335363
parsed = uuid.UUID(str(job_id))
336364
except ValueError:
@@ -339,11 +367,13 @@ def is_wire_job_id(job_id: str | None) -> TypeGuard[str]:
339367
return parsed.int != 0
340368

341369

342-
def connect_handler_ipc(pipe: str, job_id: str | None) -> _HandlerIpcConnection:
370+
def connect_handler_ipc(
371+
pipe: str, job_key: str | None, resume_version: int | None = None
372+
) -> _HandlerIpcConnection:
343373
"""Dial ``pipe`` on a dedicated loop/thread and install the sinks (see ``_HandlerIpcConnection``)."""
344-
if not is_wire_job_id(job_id):
374+
if not is_wire_job_id(job_key):
345375
raise RuntimeError(
346-
f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job id; got {job_id!r}."
376+
f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job key; got {job_key!r}."
347377
)
348378

349379
from uipath_ipc import IpcClient, NamedPipeClientTransport
@@ -369,7 +399,7 @@ async def _build() -> Any:
369399
request_timeout=_IPC_REQUEST_TIMEOUT_S,
370400
max_message_size=_MAX_MESSAGE_BYTES,
371401
)
372-
proxy = client.get_proxy(IJobInvocationCommonApi) # type: ignore[type-abstract]
402+
proxy = client.get_proxy(IPythonJobApi) # type: ignore[type-abstract]
373403
return client, proxy
374404

375405
try:
@@ -380,7 +410,7 @@ async def _build() -> Any:
380410
_stop_loop_thread(loop, thread, _SET_RESULT_TIMEOUT_S)
381411
raise
382412
# Install in the caller's context: the sinks are contextvars, read in the job's own context at teardown.
383-
handler = install_runtime_sinks(job_id, proxy, loop)
413+
handler = install_runtime_sinks(job_key, proxy, loop, resume_version)
384414
return _HandlerIpcConnection(client, loop, thread, handler)
385415

386416

@@ -392,10 +422,12 @@ async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None:
392422

393423
@contextlib.asynccontextmanager
394424
async def handler_ipc_connection(
395-
pipe: str | None, job_id: str | None
425+
pipe: str | None, job_key: str | None, resume_version: int | None = None
396426
) -> AsyncIterator[Any]:
397427
"""Connect (if ``pipe`` is set) and always disconnect on exit; yields the connection or None."""
398-
conn = connect_handler_ipc(pipe, job_id) if pipe is not None else None
428+
conn = (
429+
connect_handler_ipc(pipe, job_key, resume_version) if pipe is not None else None
430+
)
399431
try:
400432
yield conn
401433
finally:

‎packages/uipath/src/uipath/_cli/cli_server_ipc.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ async def RunJob(
106106
if request.StreamOutputOverIpc:
107107
# Never stored: a captured callback goes stale on reconnect/restart.
108108
from ._job_api import (
109-
IJobInvocationCommonApi,
109+
IPythonJobApi,
110110
clear_runtime_sinks,
111111
install_runtime_sinks,
112112
is_wire_job_id,
@@ -126,13 +126,16 @@ async def RunJob(
126126

127127
# get_callback only wraps the connection, so this cannot tell us whether the peer
128128
# actually hosts the contract; a peer that doesn't shows up as a failing send.
129-
callback = message.client.get_callback(IJobInvocationCommonApi) # type: ignore[type-abstract]
129+
callback = message.client.get_callback(IPythonJobApi) # type: ignore[type-abstract]
130130

131131
loop = asyncio.get_running_loop()
132132
job_key = request.JobKey
133+
resume_version = request.ResumeVersion
133134

134135
def _install() -> None:
135-
installed.append(install_runtime_sinks(job_key, callback, loop))
136+
installed.append(
137+
install_runtime_sinks(job_key, callback, loop, resume_version)
138+
)
136139

137140
on_run_start = _install
138141
on_run_end = clear_runtime_sinks

0 commit comments

Comments
 (0)