From 8838c75b4eae9716e87a19b1175b0b1862a6d23f Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Tue, 22 Sep 2026 09:10:51 +0200 Subject: [PATCH] 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; both are required inputs, so a lane states None on purpose rather than by omission. 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. --- packages/uipath/pyproject.toml | 2 +- packages/uipath/src/uipath/_cli/_job_api.py | 113 +++++++----- packages/uipath/src/uipath/_cli/cli_run.py | 3 +- .../uipath/src/uipath/_cli/cli_server_ipc.py | 15 +- packages/uipath/tests/cli/test_job_api.py | 163 +++++++++--------- packages/uipath/tests/cli/test_run.py | 8 +- packages/uipath/tests/cli/test_server_ipc.py | 34 ++-- packages/uipath/uv.lock | 2 +- 8 files changed, 193 insertions(+), 147 deletions(-) diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index e778925c8..010bff5d9 100644 --- a/packages/uipath/pyproject.toml +++ b/packages/uipath/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath" -version = "2.14.23" +version = "2.14.24" description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools." readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/packages/uipath/src/uipath/_cli/_job_api.py b/packages/uipath/src/uipath/_cli/_job_api.py index f288b330d..756a137bc 100644 --- a/packages/uipath/src/uipath/_cli/_job_api.py +++ b/packages/uipath/src/uipath/_cli/_job_api.py @@ -1,4 +1,4 @@ -"""The job-invocation IPC contract and the glue that routes a job's logs and result over it.""" +"""The Python job-api IPC contract and the glue that routes a job's logs and result over it.""" from __future__ import annotations @@ -55,9 +55,11 @@ class ExecutorJobStatus(IntEnum): @dataclass -class JobLogDto: +class PythonJobLogDto: """A log entry; field names are the wire keys (do not rename).""" + JobKey: str + ResumeVersion: int | None = None Message: str = "" LogLevel: int = LogLevel.INFORMATION.value @@ -74,26 +76,31 @@ class JobExecutorError: @dataclass -class JobResultDto: +class PythonJobResultDto: """The final result; field names are the wire keys (do not rename).""" - id: str = "" - status: int = ExecutorJobStatus.SUCCESSFUL.value - outputArguments: Any = None - outputArgumentsFilePath: str | None = None - info: str | None = None - error: JobExecutorError | None = None + JobKey: str + ResumeVersion: int | None = None + Status: int = ExecutorJobStatus.SUCCESSFUL.value + OutputArguments: Any = None + OutputArgumentsFilePath: str | None = None + Info: str | None = None + Error: JobExecutorError | None = None -class IJobInvocationCommonApi(ABC): - """The job-invocation contract: logs + the final result. The class name is the endpoint key.""" +class IPythonJobApi(ABC): + """The Python job-api contract: logs + the final result. The class name is the endpoint key. + + Every message names the run it belongs to (job key + resume version), so the peer can route a + pooled callback to the right job and drop a straggler from a previous resume. + """ @abstractmethod - async def SendLog(self, jobId: str, log: JobLogDto) -> None: + async def SendLog(self, log: PythonJobLogDto) -> None: """Forward one log entry.""" @abstractmethod - async def SetResult(self, jobId: str, result: JobResultDto) -> bool: + async def SetResult(self, result: PythonJobResultDto) -> bool: """Submit the final result.""" @@ -121,8 +128,11 @@ def _to_log_level(levelno: int) -> int: def _to_result_dto( - job_id: str, result: Any, output_arguments_file_path: str -) -> JobResultDto: + job_key: str, + resume_version: int | None, + result: Any, + output_arguments_file_path: str, +) -> PythonJobResultDto: error = None if result is not None and getattr(result, "error", None) is not None: category = result.error.category @@ -135,11 +145,12 @@ def _to_result_dto( ) raw_status = getattr(result, "status", None) status_key = str(getattr(raw_status, "value", raw_status) or "successful").lower() - return JobResultDto( - id=job_id, - status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value), - outputArgumentsFilePath=output_arguments_file_path, - error=error, + return PythonJobResultDto( + JobKey=job_key, + ResumeVersion=resume_version, + Status=_EXECUTOR_STATUS.get(status_key, ExecutorJobStatus.SUCCESSFUL.value), + OutputArgumentsFilePath=output_arguments_file_path, + Error=error, ) @@ -162,10 +173,15 @@ class _IpcLogHandler(logging.Handler): """Forwards each log record to the callback.""" def __init__( - self, job_id: str, callback: Any, loop: asyncio.AbstractEventLoop + self, + job_key: str, + resume_version: int | None, + callback: Any, + loop: asyncio.AbstractEventLoop, ) -> None: super().__init__() - self._job_id = job_id + self._job_key = job_key + self._resume_version = resume_version self._callback = callback self._loop = loop self._pending: set[Future[object]] = set() @@ -177,11 +193,14 @@ def emit(self, record: logging.LogRecord) -> None: _to_original_stderr(self, record) return try: - dto = JobLogDto( - Message=self.format(record), LogLevel=_to_log_level(record.levelno) + dto = PythonJobLogDto( + JobKey=self._job_key, + ResumeVersion=self._resume_version, + Message=self.format(record), + LogLevel=_to_log_level(record.levelno), ) future = asyncio.run_coroutine_threadsafe( - self._callback.SendLog(self._job_id, dto), self._loop + self._callback.SendLog(dto), self._loop ) with self._pending_lock: self._pending.add(future) @@ -226,9 +245,15 @@ async def aflush_pending(self, timeout: float = _LOG_FLUSH_TIMEOUT_S) -> None: def install_runtime_sinks( - job_id: str, callback: Any, loop: asyncio.AbstractEventLoop + job_key: str, + resume_version: int | None, + callback: Any, + loop: asyncio.AbstractEventLoop, ) -> "_IpcLogHandler | None": - """Install the log + result sinks, forwarding to ``callback`` on ``loop``. + """Install the log + result sinks for the run ``(job_key, resume_version)``, forwarding to ``callback`` on ``loop``. + + The peer routes a pooled callback by that pair exactly, so ``resume_version`` is the caller's + decision: the value it was handed, or ``None`` on a lane that has none. ``loop`` must run on a different thread than the one the sinks are invoked on, or the result ack deadlocks. Raises if this runtime has no sinks to install into. @@ -246,15 +271,15 @@ def install_runtime_sinks( "Install uipath-runtime>=0.13.5." ) from e - handler = _IpcLogHandler(job_id, callback, loop) + handler = _IpcLogHandler(job_key, resume_version, callback, loop) handler.setFormatter(logging.Formatter("%(message)s")) def _result_sink(result: Any, output_arguments_file_path: str) -> None: - dto = _to_result_dto(job_id, result, output_arguments_file_path) + dto = _to_result_dto( + job_key, resume_version, result, output_arguments_file_path + ) try: - future = asyncio.run_coroutine_threadsafe( - callback.SetResult(job_id, dto), loop - ) + future = asyncio.run_coroutine_threadsafe(callback.SetResult(dto), loop) if future.result(timeout=_SET_RESULT_TIMEOUT_S) is False: logger.error( "The handler rejected the job result (SetResult returned false)" @@ -329,21 +354,23 @@ def _shutdown(self) -> None: _stop_loop_thread(self._loop, self._thread, _SET_RESULT_TIMEOUT_S) -def is_wire_job_id(job_id: str | None) -> TypeGuard[str]: - """The peer types the job id as a Guid, and routes nothing for one it can't match.""" +def is_wire_job_key(job_key: str | None) -> TypeGuard[str]: + """The peer types the job key as a Guid, and routes nothing for one it can't match.""" try: - parsed = uuid.UUID(str(job_id)) + parsed = uuid.UUID(str(job_key)) except ValueError: return False # All-zeros is Guid's default, so it is the one unusable value a caller reaches by omission. return parsed.int != 0 -def connect_handler_ipc(pipe: str, job_id: str | None) -> _HandlerIpcConnection: +def connect_handler_ipc( + pipe: str, job_key: str | None, resume_version: int | None +) -> _HandlerIpcConnection: """Dial ``pipe`` on a dedicated loop/thread and install the sinks (see ``_HandlerIpcConnection``).""" - if not is_wire_job_id(job_id): + if not is_wire_job_key(job_key): raise RuntimeError( - f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job id; got {job_id!r}." + f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job key; got {job_key!r}." ) from uipath_ipc import IpcClient, NamedPipeClientTransport @@ -369,7 +396,7 @@ async def _build() -> Any: request_timeout=_IPC_REQUEST_TIMEOUT_S, max_message_size=_MAX_MESSAGE_BYTES, ) - proxy = client.get_proxy(IJobInvocationCommonApi) # type: ignore[type-abstract] + proxy = client.get_proxy(IPythonJobApi) # type: ignore[type-abstract] return client, proxy try: @@ -380,7 +407,7 @@ async def _build() -> Any: _stop_loop_thread(loop, thread, _SET_RESULT_TIMEOUT_S) raise # Install in the caller's context: the sinks are contextvars, read in the job's own context at teardown. - handler = install_runtime_sinks(job_id, proxy, loop) + handler = install_runtime_sinks(job_key, resume_version, proxy, loop) return _HandlerIpcConnection(client, loop, thread, handler) @@ -392,10 +419,12 @@ async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None: @contextlib.asynccontextmanager async def handler_ipc_connection( - pipe: str | None, job_id: str | None + pipe: str | None, job_key: str | None, resume_version: int | None ) -> AsyncIterator[Any]: """Connect (if ``pipe`` is set) and always disconnect on exit; yields the connection or None.""" - conn = connect_handler_ipc(pipe, job_id) if pipe is not None else None + conn = ( + connect_handler_ipc(pipe, job_key, resume_version) if pipe is not None else None + ) try: yield conn finally: diff --git a/packages/uipath/src/uipath/_cli/cli_run.py b/packages/uipath/src/uipath/_cli/cli_run.py index b94dd85ba..05b0cd350 100644 --- a/packages/uipath/src/uipath/_cli/cli_run.py +++ b/packages/uipath/src/uipath/_cli/cli_run.py @@ -231,7 +231,8 @@ async def execute() -> None: from ._job_api import handler_ipc_connection async with ( - handler_ipc_connection(handler_ipc_pipe, ctx.job_id), + # No resume version reaches this lane; the per-job pipe already names the run. + handler_ipc_connection(handler_ipc_pipe, ctx.job_id, None), ResourceOverwritesContext( lambda: read_resource_overwrites_from_file(ctx.runtime_dir) ), diff --git a/packages/uipath/src/uipath/_cli/cli_server_ipc.py b/packages/uipath/src/uipath/_cli/cli_server_ipc.py index 52f9a94bf..ffeb092ae 100644 --- a/packages/uipath/src/uipath/_cli/cli_server_ipc.py +++ b/packages/uipath/src/uipath/_cli/cli_server_ipc.py @@ -106,16 +106,16 @@ async def RunJob( if request.StreamOutputOverIpc: # Never stored: a captured callback goes stale on reconnect/restart. from ._job_api import ( - IJobInvocationCommonApi, + IPythonJobApi, clear_runtime_sinks, install_runtime_sinks, - is_wire_job_id, + is_wire_job_key, ) - if not is_wire_job_id(request.JobKey): + if not is_wire_job_key(request.JobKey): return PythonServerRunJobResult( ExitCode=1, - Error=f"StreamOutputOverIpc needs a 'JobKey' that is a job id; got {request.JobKey!r}", + Error=f"StreamOutputOverIpc needs a 'JobKey' that is a job key (Guid); got {request.JobKey!r}", ) if message is None or message.client is None: @@ -126,13 +126,16 @@ async def RunJob( # get_callback only wraps the connection, so this cannot tell us whether the peer # actually hosts the contract; a peer that doesn't shows up as a failing send. - callback = message.client.get_callback(IJobInvocationCommonApi) # type: ignore[type-abstract] + callback = message.client.get_callback(IPythonJobApi) # type: ignore[type-abstract] loop = asyncio.get_running_loop() job_key = request.JobKey + resume_version = request.ResumeVersion def _install() -> None: - installed.append(install_runtime_sinks(job_key, callback, loop)) + installed.append( + install_runtime_sinks(job_key, resume_version, callback, loop) + ) on_run_start = _install on_run_end = clear_runtime_sinks diff --git a/packages/uipath/tests/cli/test_job_api.py b/packages/uipath/tests/cli/test_job_api.py index 46a29ff34..9ac708b82 100644 --- a/packages/uipath/tests/cli/test_job_api.py +++ b/packages/uipath/tests/cli/test_job_api.py @@ -40,21 +40,21 @@ class _Result: status = UiPathRuntimeStatus.FAULTED error = _Error() - dto = _job_api._to_result_dto("job-1", _Result(), "out.args") + dto = _job_api._to_result_dto("job-1", 3, _Result(), "out.args") - assert dto.id == "job-1" - assert dto.status == _job_api.ExecutorJobStatus.FAULTED.value - assert dto.outputArgumentsFilePath == "out.args" - assert dto.outputArguments is None - assert dto.error is not None + assert (dto.JobKey, dto.ResumeVersion) == ("job-1", 3) + assert dto.Status == _job_api.ExecutorJobStatus.FAULTED.value + assert dto.OutputArgumentsFilePath == "out.args" + assert dto.OutputArguments is None + assert dto.Error is not None # Every field, so a swapped Title/Detail (a stack trace shown as the error's title in the # job's failure record) cannot pass. assert ( - dto.error.Code, - dto.error.Title, - dto.error.Detail, - dto.error.Category, - dto.error.Status, + dto.Error.Code, + dto.Error.Title, + dto.Error.Detail, + dto.Error.Category, + dto.Error.Status, ) == ("BOOM", "It broke", "stack", "User", 404) @@ -63,9 +63,10 @@ class _Result: status = UiPathRuntimeStatus.SUCCESSFUL error = None - dto = _job_api._to_result_dto("j", _Result(), "p.args") - assert dto.status == _job_api.ExecutorJobStatus.SUCCESSFUL.value - assert dto.error is None + dto = _job_api._to_result_dto("j", None, _Result(), "p.args") + assert dto.Status == _job_api.ExecutorJobStatus.SUCCESSFUL.value + assert dto.ResumeVersion is None + assert dto.Error is None def test_to_result_dto_maps_suspended(): @@ -77,8 +78,8 @@ class _Result: status = UiPathRuntimeStatus.SUSPENDED error = None - dto = _job_api._to_result_dto("j", _Result(), "p.args") - assert dto.status == _job_api.ExecutorJobStatus.SUSPENDED.value + dto = _job_api._to_result_dto("j", None, _Result(), "p.args") + assert dto.Status == _job_api.ExecutorJobStatus.SUSPENDED.value def test_to_log_level_maps_python_levels_to_wire_values(): @@ -95,24 +96,32 @@ def test_to_log_level_maps_python_levels_to_wire_values(): def test_dto_wire_key_sets_are_pinned(): """Pin each DTO's on-wire JSON keys so an accidental rename is caught on this side. - Guards our half of the wire contract: JobResultDto is camelCase, JobLogDto / JobExecutorError - are PascalCase. + Guards our half of the wire contract: every DTO is PascalCase, matching the peer's property + names (its base-class camelCase JSON names are matched case-insensitively). """ serialization = pytest.importorskip("uipath_ipc.wire.serialization") to_wire = serialization.to_wire result_keys = set( - to_wire(_job_api.JobResultDto(id="j", outputArgumentsFilePath="p.args")) + to_wire( + _job_api.PythonJobResultDto(JobKey="j", OutputArgumentsFilePath="p.args") + ) ) assert result_keys == { - "id", - "status", - "outputArguments", - "outputArgumentsFilePath", - "info", - "error", + "JobKey", + "ResumeVersion", + "Status", + "OutputArguments", + "OutputArgumentsFilePath", + "Info", + "Error", + } + assert set(to_wire(_job_api.PythonJobLogDto(JobKey="j", Message="m"))) == { + "JobKey", + "ResumeVersion", + "Message", + "LogLevel", } - assert set(to_wire(_job_api.JobLogDto(Message="m"))) == {"Message", "LogLevel"} assert set(to_wire(_job_api.JobExecutorError(Code="c"))) == { "Code", "Title", @@ -152,30 +161,30 @@ def __getitem__(self, key: str) -> Any: def test_install_wires_log_handler_and_result_sink(monkeypatch): captured = _isolated_output_sinks(monkeypatch) - logs: list[tuple[str, Any]] = [] - results: list[tuple[str, Any]] = [] + logs: list[Any] = [] + results: list[Any] = [] class _Callback: - async def SendLog(self, jid: str, dto: Any) -> None: - logs.append((jid, dto)) + async def SendLog(self, dto: Any) -> None: + logs.append(dto) - async def SetResult(self, jid: str, dto: Any) -> bool: - results.append((jid, dto)) + async def SetResult(self, dto: Any) -> bool: + results.append(dto) return True async def scenario() -> None: loop = asyncio.get_running_loop() - _job_api.install_runtime_sinks("job-7", _Callback(), loop) + _job_api.install_runtime_sinks("job-7", 2, _Callback(), loop) - # The log handler forwards each record, tagged with the job id. + # The log handler forwards each record, tagged with the run it belongs to. handler = captured["handler"] handler.emit( logging.LogRecord("n", logging.WARNING, "p", 1, "hi %s", ("there",), None) ) await handler.aflush_pending() - assert logs[0][0] == "job-7" - assert logs[0][1].Message == "hi there" - assert logs[0][1].LogLevel == _job_api.LogLevel.WARNING.value + assert (logs[0].JobKey, logs[0].ResumeVersion) == ("job-7", 2) + assert logs[0].Message == "hi there" + assert logs[0].LogLevel == _job_api.LogLevel.WARNING.value # The result sink maps the result and calls SetResult, off a worker thread, for the ack. class _Result: @@ -184,8 +193,8 @@ class _Result: sink = captured["sink"] await asyncio.to_thread(sink, _Result(), "out.args") - assert results[0][0] == "job-7" - assert results[0][1].outputArgumentsFilePath == "out.args" + assert (results[0].JobKey, results[0].ResumeVersion) == ("job-7", 2) + assert results[0].OutputArgumentsFilePath == "out.args" asyncio.run(scenario()) @@ -195,12 +204,12 @@ def test_an_unformattable_record_does_not_escape_emit(monkeypatch): captured = _isolated_output_sinks(monkeypatch) class _Callback: - async def SendLog(self, jid: str, log: Any) -> None: + async def SendLog(self, log: Any) -> None: return None loop = asyncio.new_event_loop() try: - _job_api.install_runtime_sinks(JOB_ID, _Callback(), loop) + _job_api.install_runtime_sinks(JOB_ID, None, _Callback(), loop) handler = captured["handler"] handled: list[Any] = [] monkeypatch.setattr(handler, "handleError", handled.append) @@ -218,8 +227,8 @@ def test_transport_and_self_logs_never_ride_the_ipc_channel(monkeypatch): forwarded = threading.Event() class _Callback: - async def SendLog(self, jid: str, dto: Any) -> None: - sent.append((jid, dto)) + async def SendLog(self, dto: Any) -> None: + sent.append(dto) forwarded.set() buf = io.StringIO() @@ -229,7 +238,7 @@ async def SendLog(self, jid: str, dto: Any) -> None: thread = threading.Thread(target=loop.run_forever, daemon=True) thread.start() try: - _job_api.install_runtime_sinks("job-6", _Callback(), loop) + _job_api.install_runtime_sinks("job-6", None, _Callback(), loop) handler = captured["handler"] for name in ("uipath_ipc.client.connection", _job_api.__name__): handler.emit( @@ -246,7 +255,7 @@ async def SendLog(self, jid: str, dto: Any) -> None: thread.join(timeout=5) loop.close() - assert [dto.Message for _, dto in sent] == ["job line"] + assert [dto.Message for dto in sent] == ["job line"] assert buf.getvalue().count("internal chatter") == 2 @@ -254,7 +263,7 @@ def test_rejected_result_is_reported(monkeypatch): captured = _isolated_output_sinks(monkeypatch) class _Callback: - async def SetResult(self, jid: str, dto: Any) -> bool: + async def SetResult(self, dto: Any) -> bool: return False class _Result: @@ -277,7 +286,7 @@ def emit(self, record: logging.LogRecord) -> None: thread = threading.Thread(target=loop.run_forever, daemon=True) thread.start() try: - _job_api.install_runtime_sinks("job-8", _Callback(), loop) + _job_api.install_runtime_sinks("job-8", None, _Callback(), loop) captured["sink"](_Result(), "out.args") finally: loop.call_soon_threadsafe(loop.stop) @@ -320,7 +329,7 @@ def test_pending_log_sends_are_flushed_before_teardown(monkeypatch): landed: list[str] = [] class _Callback: - async def SendLog(self, jid: str, dto: Any) -> None: + async def SendLog(self, dto: Any) -> None: await asyncio.sleep(0.2) landed.append(dto.Message) @@ -331,7 +340,7 @@ async def aclose(self) -> None: loop = asyncio.new_event_loop() thread = threading.Thread(target=loop.run_forever, daemon=True) thread.start() - handler = _job_api.install_runtime_sinks("job-7", _Callback(), loop) + handler = _job_api.install_runtime_sinks("job-7", None, _Callback(), loop) assert handler is not None handler.emit(logging.LogRecord("j", logging.INFO, "p", 1, "tail line", (), None)) @@ -345,7 +354,7 @@ def test_result_delivery_failure_is_swallowed_and_logged(monkeypatch): captured = _isolated_output_sinks(monkeypatch) class _Callback: - async def SetResult(self, jid: str, dto: Any) -> bool: + async def SetResult(self, dto: Any) -> bool: raise RuntimeError("handler said no") class _Result: @@ -371,7 +380,7 @@ def emit(self, record: logging.LogRecord) -> None: thread = threading.Thread(target=loop.run_forever, daemon=True) thread.start() try: - _job_api.install_runtime_sinks("job-9", _Callback(), loop) + _job_api.install_runtime_sinks("job-9", None, _Callback(), loop) sink = captured["sink"] sink(_Result(), "out.args") # must not raise finally: @@ -396,11 +405,11 @@ def test_connect_installs_sinks_and_disconnect_clears(monkeypatch): pytest.importorskip("uipath_ipc") captured = _isolated_output_sinks(monkeypatch) - class _Api(_job_api.IJobInvocationCommonApi): - async def SendLog(self, jobId: str, log: Any) -> None: + class _Api(_job_api.IPythonJobApi): + async def SendLog(self, log: Any) -> None: return None - async def SetResult(self, jobId: str, result: Any) -> bool: + async def SetResult(self, result: Any) -> bool: return True pipe = _unique_jobapi_pipe() @@ -410,7 +419,7 @@ async def SetResult(self, jobId: str, result: Any) -> bool: async def scenario() -> None: # The context manager, not the two halves: cli_run.py only ever uses this, and a # wiring mistake inside it would leave every assertion below untouched. - async with _job_api.handler_ipc_connection(pipe, JOB_ID): + async with _job_api.handler_ipc_connection(pipe, JOB_ID, None): assert captured["handler"] is not None assert captured["sink"] is not None @@ -438,7 +447,7 @@ def live() -> int: before = live() with pytest.raises(RuntimeError, match="Could not reach the handler IPC pipe"): - _job_api.connect_handler_ipc("uipath-jobapi-does-not-exist-12345", JOB_ID) + _job_api.connect_handler_ipc("uipath-jobapi-does-not-exist-12345", JOB_ID, None) assert live() == before @@ -450,14 +459,14 @@ def test_a_broken_log_channel_is_reported_once_to_stderr(monkeypatch): monkeypatch.setattr(sys, "__stderr__", buf) class _Callback: - async def SendLog(self, job_id: str, log: Any) -> None: + async def SendLog(self, log: Any) -> None: raise RuntimeError("pipe is gone") loop = asyncio.new_event_loop() thread = threading.Thread(target=loop.run_forever, daemon=True) thread.start() try: - handler = _job_api.install_runtime_sinks(JOB_ID, _Callback(), loop) + handler = _job_api.install_runtime_sinks(JOB_ID, None, _Callback(), loop) assert handler is not None for i in range(4): handler.emit( @@ -483,7 +492,7 @@ def test_a_cancelled_send_is_reported_like_any_other_failure(monkeypatch): loop = asyncio.new_event_loop() try: - handler = _job_api._IpcLogHandler(JOB_ID, object(), loop) + handler = _job_api._IpcLogHandler(JOB_ID, None, object(), loop) cancelled: Future[object] = Future() assert cancelled.cancel() @@ -499,7 +508,7 @@ def test_missing_output_sinks_fails_loudly(monkeypatch): monkeypatch.setitem(sys.modules, "uipath.runtime.output_sinks", None) with pytest.raises(RuntimeError, match="uipath-runtime"): - _job_api.install_runtime_sinks(JOB_ID, object(), asyncio.new_event_loop()) + _job_api.install_runtime_sinks(JOB_ID, None, object(), asyncio.new_event_loop()) def test_the_frame_cap_matches_the_dotnet_peer(): @@ -535,14 +544,14 @@ def _live_ipc_threads() -> int: def test_connect_without_a_real_job_id_fails_fast(job_id): before = _live_ipc_threads() with pytest.raises(RuntimeError, match="UIPATH_JOB_KEY"): - _job_api.connect_handler_ipc("pipe", job_id) + _job_api.connect_handler_ipc("pipe", job_id, None) assert _live_ipc_threads() == before @pytest.mark.parametrize("job_id", [None, "", "job-1"]) async def test_handler_ipc_connection_without_a_real_job_id_fails_fast(job_id): with pytest.raises(RuntimeError, match="UIPATH_JOB_KEY"): - async with _job_api.handler_ipc_connection("pipe", job_id): + async with _job_api.handler_ipc_connection("pipe", job_id, None): pass @@ -556,7 +565,7 @@ def _unique_jobapi_pipe() -> str: def _serve_jobapi_in_background(pipe: str, api: Any): - """Host ``api`` as IJobInvocationCommonApi on ``pipe`` in a daemon thread; return a stop() callable. + """Host ``api`` as IPythonJobApi on ``pipe`` in a daemon thread; return a stop() callable. The server runs on its OWN loop/thread so it can keep accepting while the test thread is blocked inside the (synchronous) result sink — the whole point of the regression below. @@ -570,7 +579,7 @@ def _serve_jobapi_in_background(pipe: str, api: Any): async def _serve() -> None: server = IpcServer( transport=NamedPipeServerTransport(pipe), - services={_job_api.IJobInvocationCommonApi: api}, + services={_job_api.IPythonJobApi: api}, request_timeout=None, ) holder["server"] = server @@ -619,14 +628,14 @@ def test_result_sink_delivers_when_invoked_on_the_caller_loop_thread(monkeypatch captured = _isolated_output_sinks(monkeypatch) received: dict[str, Any] = {} - class _Api(_job_api.IJobInvocationCommonApi): - async def SendLog(self, jobId: str, log: Any) -> None: - received.setdefault("logs", []).append((jobId, log)) + class _Api(_job_api.IPythonJobApi): + async def SendLog(self, log: Any) -> None: + received.setdefault("logs", []).append(log) - async def SetResult(self, jobId: str, result: Any) -> bool: + async def SetResult(self, result: Any) -> bool: # The server deserializes against the contract's typed signature, so result arrives as a - # real JobResultDto (this also exercises the wire round-trip of the DTO). - received["result"] = (jobId, result) + # real PythonJobResultDto (this also exercises the wire round-trip of the DTO). + received["result"] = result return True class _Result: @@ -638,7 +647,7 @@ class _Result: try: async def scenario() -> None: - conn = _job_api.connect_handler_ipc(pipe, JOB_ID_2) + conn = _job_api.connect_handler_ipc(pipe, JOB_ID_2, 1) # A log line has to survive the round trip too, not just the result. captured["handler"].emit( logging.LogRecord( @@ -669,15 +678,15 @@ async def scenario() -> None: # The log half of the contract, asserted on the wire rather than against a fake. assert "logs" in received, "SendLog never arrived over the pipe" - log_job_id, entry = received["logs"][0] - assert log_job_id == JOB_ID_2 + entry = received["logs"][0] + assert (entry.JobKey, entry.ResumeVersion) == (JOB_ID_2, 1) assert entry.Message == "over ipc" assert entry.LogLevel == _job_api.LogLevel.WARNING.value assert "result" in received, ( "SetResult never arrived — the result sink deadlocked/timed out" ) - job_id, dto = received["result"] - assert job_id == JOB_ID_2 - assert dto.outputArgumentsFilePath == "out.args" - assert dto.status == _job_api.ExecutorJobStatus.SUCCESSFUL.value + dto = received["result"] + assert (dto.JobKey, dto.ResumeVersion) == (JOB_ID_2, 1) + assert dto.OutputArgumentsFilePath == "out.args" + assert dto.Status == _job_api.ExecutorJobStatus.SUCCESSFUL.value diff --git a/packages/uipath/tests/cli/test_run.py b/packages/uipath/tests/cli/test_run.py index a09b4ac33..a47e5a4d7 100644 --- a/packages/uipath/tests/cli/test_run.py +++ b/packages/uipath/tests/cli/test_run.py @@ -235,7 +235,7 @@ def test_the_job_runs_inside_the_open_connection( monkeypatch.setenv("UIPATH_JOB_KEY", job_key) events: list[str] = [] - seen: list[tuple[str, str]] = [] + seen: list[tuple[str, str, int | None]] = [] snapshots: list[object] = [] from uipath.runtime import context as runtime_context @@ -244,8 +244,8 @@ def test_the_job_runs_inside_the_open_connection( sentinel = logging.Handler() @asynccontextmanager - async def _fake_connection(pipe, job_id): - seen.append((pipe, job_id)) + async def _fake_connection(pipe, job_key, resume_version): + seen.append((pipe, job_key, resume_version)) events.append("open") # Install for real: the ordering that matters is against the runtime context's # one-shot snapshot, not against runtime.execute. @@ -303,7 +303,7 @@ async def _execute(*a, **k): # The whole point: the connection must still be open when the job runs. assert events == ["open", "run", "close"], events # And it must be told which job, not None. - assert seen == [("some-pipe", job_key)] + assert seen == [("some-pipe", job_key, None)] # The connection must be open BEFORE the runtime context snapshots the sinks: move it # inside and __enter__ would capture None, losing every log line and the result. assert snapshots == [sentinel], ( diff --git a/packages/uipath/tests/cli/test_server_ipc.py b/packages/uipath/tests/cli/test_server_ipc.py index c48e76354..8c57cd3ef 100644 --- a/packages/uipath/tests/cli/test_server_ipc.py +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -30,7 +30,7 @@ ) from uipath._cli import _server_core -from uipath._cli._job_api import IJobInvocationCommonApi +from uipath._cli._job_api import IPythonJobApi from uipath._cli.cli_server import ( IPythonRuntimeServer, PythonRuntimeService, @@ -393,7 +393,7 @@ def test_runjob_without_a_real_job_key_fails_loudly(self, monkeypatch, job_key): monkeypatch.setattr( _job_api, "install_runtime_sinks", - lambda jid, cb, loop: events.append(("install", jid)), + lambda jid, rv, cb, loop: events.append(("install", jid)), ) async def _fake_run(*args: Any, **kwargs: Any) -> dict[str, Any]: @@ -450,14 +450,16 @@ def test_pooled_streaming_works_over_a_real_pipe(self, monkeypatch): """No hand-fed ``message``: the dispatcher must inject it from the contract.""" from uipath._cli import _job_api, cli_server_ipc - installed: list[Any] = [] + installed: list[tuple[str, int | None]] = [] real_install = _job_api.install_runtime_sinks - def _record(jid, cb, loop): - installed.append(jid) + def _record( + jid: str, rv: int | None, cb: Any, loop: asyncio.AbstractEventLoop + ) -> Any: + installed.append((jid, rv)) # Install for real: patching this away would leave the callback unused, so nothing # would notice if RunJob asked the peer for the wrong contract. - return real_install(jid, cb, loop) + return real_install(jid, rv, cb, loop) monkeypatch.setattr(_job_api, "install_runtime_sinks", _record) @@ -484,10 +486,10 @@ async def _fake_run(cmd, args, env, wd, on_run_start=None, on_run_end=None): delivered: list[tuple[str, Any]] = [] class _Callback: - async def SendLog(self, job_id: str, log: Any) -> None: - delivered.append((job_id, log)) + async def SendLog(self, log: Any) -> None: + delivered.append((log.JobKey, log)) - async def SetResult(self, job_id: str, result: Any) -> bool: + async def SetResult(self, result: Any) -> bool: return True pipe = _unique_pipe() @@ -496,7 +498,7 @@ async def SetResult(self, job_id: str, result: Any) -> bool: async def scenario() -> Any: client = IpcClient( transport=NamedPipeClientTransport(pipe), - callbacks={IJobInvocationCommonApi: _Callback()}, + callbacks={IPythonJobApi: _Callback()}, ) try: proxy = client.get_proxy(IPythonRuntimeServer) # type: ignore[type-abstract] @@ -505,6 +507,7 @@ async def scenario() -> Any: Any, { "JobKey": JOB_ID, + "ResumeVersion": 2, "Command": "run", "Args": [], "StreamOutputOverIpc": True, @@ -517,11 +520,12 @@ async def scenario() -> Any: result = asyncio.run(scenario()) assert result.Error is None, result.Error assert result.ExitCode == 0 - assert installed == [JOB_ID] + assert installed == [(JOB_ID, 2)] # The contract passed to get_callback IS the endpoint key on the wire: ask for the wrong # one and every send is addressed to something the peer does not host. assert len(delivered) == 1, "no log line crossed the pooled callback" - assert delivered[0][0] == JOB_ID + # The peer routes by (JobKey, ResumeVersion) exactly; a null here would miss a resumed job. + assert (delivered[0][1].JobKey, delivered[0][1].ResumeVersion) == (JOB_ID, 2) assert delivered[0][1].Message == "pooled line" def test_runjob_installs_the_sinks_from_the_request_callback(self, monkeypatch): @@ -532,7 +536,7 @@ def test_runjob_installs_the_sinks_from_the_request_callback(self, monkeypatch): monkeypatch.setattr( _job_api, "install_runtime_sinks", - lambda jid, cb, loop: events.append(("install", jid, cb)), + lambda jid, rv, cb, loop: events.append(("install", jid, cb)), ) monkeypatch.setattr( _job_api, "clear_runtime_sinks", lambda: events.append(("clear",)) @@ -574,7 +578,7 @@ async def aflush_pending(self, *a: Any, **k: Any) -> None: events.append("flush") monkeypatch.setattr( - _job_api, "install_runtime_sinks", lambda jid, cb, loop: _Handler() + _job_api, "install_runtime_sinks", lambda jid, rv, cb, loop: _Handler() ) monkeypatch.setattr(_job_api, "clear_runtime_sinks", lambda: None) @@ -611,7 +615,7 @@ def test_runjob_skips_sinks_when_not_opted_in(self, monkeypatch): monkeypatch.setattr( _job_api, "install_runtime_sinks", - lambda jid, cb, loop: events.append(("install",)), + lambda jid, rv, cb, loop: events.append(("install",)), ) monkeypatch.setattr( _job_api, "clear_runtime_sinks", lambda: events.append(("clear",)) diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index cd8ee6671..371696ba7 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2599,7 +2599,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.14.23" +version = "2.14.24" source = { editable = "." } dependencies = [ { name = "applicationinsights" },