fix(cli): make the run identity on the job-api DTOs explicit and pin the pooled resume version

JobKey has no default any more (the peer types it Guid; an empty string would be rejected), and
resume_version is a required parameter on the sink installer and the connection helpers, so a lane
states None on purpose rather than by omission. is_wire_job_id is is_wire_job_key. The pooled
real-pipe test now sends ResumeVersion and asserts it on the received log.

Author: eduard-dumitru

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

@@ -58,7 +58,7 @@ class ExecutorJobStatus(IntEnum):
 class PythonJobLogDto:
     """A log entry; field names are the wire keys (do not rename)."""
 
-    JobKey: str = ""
+    JobKey: str
     ResumeVersion: int | None = None
     Message: str = ""
     LogLevel: int = LogLevel.INFORMATION.value
@@ -79,7 +79,7 @@ class JobExecutorError:
 class PythonJobResultDto:
     """The final result; field names are the wire keys (do not rename)."""
 
-    JobKey: str = ""
+    JobKey: str
     ResumeVersion: int | None = None
     Status: int = ExecutorJobStatus.SUCCESSFUL.value
     OutputArguments: Any = None
@@ -175,9 +175,9 @@ class _IpcLogHandler(logging.Handler):
     def __init__(
         self,
         job_key: str,
+        resume_version: int | None,
         callback: Any,
         loop: asyncio.AbstractEventLoop,
-        resume_version: int | None = None,
     ) -> None:
         super().__init__()
         self._job_key = job_key
@@ -246,11 +246,14 @@ async def aflush_pending(self, timeout: float = _LOG_FLUSH_TIMEOUT_S) -> None:
 
 def install_runtime_sinks(
     job_key: str,
+    resume_version: int | None,
     callback: Any,
     loop: asyncio.AbstractEventLoop,
-    resume_version: int | None = None,
 ) -> "_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.
@@ -268,7 +271,7 @@ def install_runtime_sinks(
             "Install uipath-runtime>=0.13.5."
         ) from e
 
-    handler = _IpcLogHandler(job_key, callback, loop, resume_version)
+    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:
@@ -351,21 +354,21 @@ 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]:
+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_key: str | None, resume_version: int | None = None
+    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_key):
+    if not is_wire_job_key(job_key):
         raise RuntimeError(
             f"--handler-ipc-pipe needs UIPATH_JOB_KEY to be a job key; got {job_key!r}."
         )
@@ -404,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_key, proxy, loop, resume_version)
+    handler = install_runtime_sinks(job_key, resume_version, proxy, loop)
     return _HandlerIpcConnection(client, loop, thread, handler)
 
 
@@ -416,7 +419,7 @@ async def disconnect_handler_ipc(conn: _HandlerIpcConnection) -> None:
 
 @contextlib.asynccontextmanager
 async def handler_ipc_connection(
-    pipe: str | None, job_key: str | None, resume_version: int | None = 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 = (

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)
                     ),

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

@@ -109,13 +109,13 @@ async def RunJob(
                 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:
@@ -134,7 +134,7 @@ async def RunJob(
 
             def _install() -> None:
                 installed.append(
-                    install_runtime_sinks(job_key, callback, loop, resume_version)
+                    install_runtime_sinks(job_key, resume_version, callback, loop)
                 )
 
             on_run_start = _install

packages/uipath/tests/cli/test_job_api.py

@@ -116,7 +116,7 @@ def test_dto_wire_key_sets_are_pinned():
         "Info",
         "Error",
     }
-    assert set(to_wire(_job_api.PythonJobLogDto(Message="m"))) == {
+    assert set(to_wire(_job_api.PythonJobLogDto(JobKey="j", Message="m"))) == {
         "JobKey",
         "ResumeVersion",
         "Message",
@@ -174,7 +174,7 @@ async def SetResult(self, dto: Any) -> bool:
 
     async def scenario() -> None:
         loop = asyncio.get_running_loop()
-        _job_api.install_runtime_sinks("job-7", _Callback(), loop, resume_version=2)
+        _job_api.install_runtime_sinks("job-7", 2, _Callback(), loop)
 
         # The log handler forwards each record, tagged with the run it belongs to.
         handler = captured["handler"]
@@ -209,7 +209,7 @@ async def SendLog(self, log: Any) -> 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)
@@ -238,7 +238,7 @@ async def SendLog(self, 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(
@@ -286,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)
@@ -340,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))
 
@@ -380,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:
@@ -419,7 +419,7 @@ async def SetResult(self, 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
 
@@ -447,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
 
@@ -466,7 +466,7 @@ async def SendLog(self, log: Any) -> None:
     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(
@@ -492,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()
 
@@ -508,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():
@@ -544,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
 
 
@@ -647,7 +647,7 @@ class _Result:
     try:
 
         async def scenario() -> None:
-            conn = _job_api.connect_handler_ipc(pipe, JOB_ID_2, resume_version=1)
+            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(

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], (

packages/uipath/tests/cli/test_server_ipc.py

@@ -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, rv=None: 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, rv=None):
-            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, rv)
+            return real_install(jid, rv, cb, loop)
 
         monkeypatch.setattr(_job_api, "install_runtime_sinks", _record)
 
@@ -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, rv=None: 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, rv=None: _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, rv=None: events.append(("install",)),
+            lambda jid, rv, cb, loop: events.append(("install",)),
         )
         monkeypatch.setattr(
             _job_api, "clear_runtime_sinks", lambda: events.append(("clear",))