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" },