feat(cli): really stop what uipath server is running

run/debug/eval drive their own event loop on the server's worker thread. They now publish it through run_execution_loop to an ExecutionControl, so a stop cancels the execution's root task and the runtime unwinds cooperatively, still writing its result. stop_execution backs both IPC StopJob and the new POST /jobs/{key}/stop. It cancels the root task, waits a grace period, cancels every task on the execution's loop, and answers False if it is still running, because it is blocked in a call that only ending the process can interrupt. forceStop shortens the waits. A queued execution is dropped before it runs. A stop that targets another resume version, or an unknown key, answers True: that run is not running. The lock, env and cwd are no longer handed on while the execution's thread still runs. A cancelled caller (a dropped IPC connection, a shutdown) stops the execution and re-raises only once the thread has exited, and the job scope's teardown completes before the env is restored. A stopped execution ends with exit code 143 and "Stopped on request"; a CancelledError it raised on its own is reported as an unexpected failure. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
robert-ursu · Sep 24, 2026 · Sep 24, 2026 · Oct 5, 2026 · Oct 5, 2026 · c9b307d5ef21b980be366325edc909dda90dcb5e
commit c9b307d5ef21b980be366325edc909dda90dcb5e
diff --git a/packages/uipath/src/uipath/_cli/_execution_control.py b/packages/uipath/src/uipath/_cli/_execution_control.py
@@ -0,0 +1,99 @@
+"""Lets the server cancel the execution that runs on its worker thread.
+
+run/debug/eval drive their own event loop inside the ``asyncio.to_thread`` worker. That
+loop is the only place a cancellation can land, so the command publishes it here and
+the server reaches it through ``loop.call_soon_threadsafe``.
+
+Kept import-light: ``_cli/__init__.py`` defers heavy imports.
+"""
+
+import asyncio
+import contextvars
+import threading
+from typing import Any
+
+CURRENT_EXECUTION_CONTROL: contextvars.ContextVar["ExecutionControl | None"] = (
+    contextvars.ContextVar("uipath_current_execution_control", default=None)
+)
+
+
+class ExecutionControl:
+    """Handle on one execution's event loop, shared by the server and its thread."""
+
+    def __init__(self) -> None:
+        """Create an unbound control; the execution binds its loop once it has one."""
+        self.cancel_requested = False
+        self._delivered = False
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._task: "asyncio.Task[Any] | None" = None
+        self._sync = threading.Lock()
+
+    def bind(self, loop: asyncio.AbstractEventLoop, task: "asyncio.Task[Any]") -> None:
+        """Publish the execution's loop and root task (execution thread)."""
+        with self._sync:
+            self._loop, self._task = loop, task
+            pending = self.cancel_requested
+        if pending:
+            self.cancel()
+
+    def unbind(self) -> None:
+        """Withdraw the loop once the root task is done (execution thread)."""
+        with self._sync:
+            self._loop = self._task = None
+
+    def cancel(self) -> None:
+        """Cancel the execution's root task, at most once.
+
+        A second delivery would land inside the runtime's cleanup ``finally`` blocks,
+        the ones that write ``output.json``, and abort them. Before the execution has a
+        loop, the request is recorded and applied by ``bind``.
+        """
+        with self._sync:
+            self.cancel_requested = True
+            if self._delivered or self._loop is None or self._task is None:
+                return
+            loop, task = self._loop, self._task
+            self._delivered = True
+        try:
+            loop.call_soon_threadsafe(task.cancel)
+        except RuntimeError:
+            with self._sync:
+                self._delivered = False
+
+    def cancel_all(self) -> None:
+        """Cancel every task on the execution's loop, giving up on a clean cleanup."""
+        with self._sync:
+            self.cancel_requested = True
+            loop = self._loop
+        if loop is None:
+            return
+
+        def _sweep() -> None:
+            for task in asyncio.all_tasks(loop):
+                task.cancel()
+
+        try:
+            loop.call_soon_threadsafe(_sweep)
+        except RuntimeError:
+            pass
+
+
+def run_execution_loop(coro: Any) -> Any:
+    """``asyncio.run`` that publishes its loop and root task to the execution's control.
+
+    With no control in scope (``uipath run`` from a terminal) this is ``asyncio.run``.
+    The loop is withdrawn before the runner closes, so a sweep can never cancel the
+    runner's own wait on the threads the execution started.
+    """
+    control = CURRENT_EXECUTION_CONTROL.get()
+    if control is None:
+        return asyncio.run(coro)
+
+    with asyncio.Runner() as runner:
+        loop = runner.get_loop()
+        task = loop.create_task(coro, context=contextvars.copy_context())
+        control.bind(loop, task)
+        try:
+            return loop.run_until_complete(task)
+        finally:
+            control.unbind()