fix(py): kill a worker that stops itself instead of freezing the host

pkg-py/src/commons/_execution/_backend.py

@@ -16,6 +16,7 @@
 import shutil
 import signal
 import tempfile
+import threading
 from collections.abc import Coroutine
 from pathlib import Path
 from typing import Any, Literal, Protocol, runtime_checkable
@@ -234,6 +235,12 @@ def __init__(
         self._closing: asyncio.Task[None] | None = None
         self._exit = asyncio.ensure_future(process.wait())
         self._exit.add_done_callback(self._kill_group)
+        threading.Thread(
+            target=_kill_when_stopped,
+            args=(process.pid,),
+            name=f"commons-worker-stop-{process.pid}",
+            daemon=True,
+        ).start()
 
     @property
     def stdin(self) -> asyncio.StreamWriter:
@@ -301,6 +308,37 @@ async def _finish(self) -> None:
         shutil.rmtree(self.scratch, ignore_errors=True)
 
 
+def _kill_when_stopped(pid: int) -> None:
+    """Kill the worker's process group if the worker is stopped.
+
+    Waits until the worker stops or exits, and returns without acting if
+    it exits. Model code can stop its own process with SIGSTOP (nothing
+    else has a reason to). On macOS, Python 3.14's asyncio mistakes a
+    stopped child for an exited one, then blocks the event loop in
+    ``waitpid()`` until the child really exits, which freezes the host.
+    Killing the worker lets that ``waitpid()`` return at once. This runs
+    wherever ``os.waitid`` exists, so the driver reports a stopped worker
+    as a crash on every such platform.
+
+    ``WNOWAIT`` leaves the exit status for asyncio to collect. Without
+    ``os.waitid`` (macOS before Python 3.13), a stopped worker is not
+    killed here, and the call fails when its timeout runs out.
+    """
+    if not hasattr(os, "waitid"):
+        return
+    try:
+        info = os.waitid(os.P_PID, pid, os.WEXITED | os.WSTOPPED | os.WNOWAIT)
+    except ChildProcessError:
+        return
+    if info is None or info.si_code != os.CLD_STOPPED:
+        return
+    with contextlib.suppress(ProcessLookupError, PermissionError):
+        os.killpg(pid, signal.SIGKILL)
+        return
+    with contextlib.suppress(ProcessLookupError):
+        os.kill(pid, signal.SIGKILL)
+
+
 def _signal_tree(process: asyncio.subprocess.Process, sig: signal.Signals) -> None:
     """Signal the child's whole process group, falling back to the child alone.
 

pkg-py/tests/test_execution_driver.py

@@ -10,7 +10,6 @@
 import asyncio
 import os
 import signal
-import sys
 import time
 from datetime import date
 from typing import Annotated
@@ -22,7 +21,7 @@
 from commons import measure, semantic_layer
 from commons._execution._backend import WORKER_SCRIPT, LocalBackend, LocalSession
 from commons._execution._driver import Failure, Worker
-from commons._execution._protocol import Error, Result, Text
+from commons._execution._protocol import Error, Ready, Result, Text, encode_message
 from commons._execution._sandbox import protection_mode
 from commons._handles import HandleStore
 from commons._measures import Injected
@@ -355,18 +354,37 @@ async def test_a_child_ignoring_sigterm_does_not_outlive_the_close():
     )
 
 
-@pytest.mark.skipif(
-    sys.platform == "darwin" and sys.version_info >= (3, 14),
-    reason="asyncio on 3.14 reads macOS's waitid() report of a stopped child "
-    "as an exit and blocks the event loop in waitpid() until it really exits",
-)
-async def test_a_worker_that_stops_reading_fails_the_call_instead_of_hanging():
-    async with make_worker(call_timeout=0.5) as worker:
+@pytest.mark.skipif(not hasattr(os, "waitid"), reason="needs os.waitid")
+async def test_a_worker_that_stops_itself_is_killed_and_reported_as_a_crash():
+    async with make_worker(call_timeout=30) as worker:
         await worker.run("1")
         process = process_of(worker)
-        # A stopped worker cannot drain its stdin; a call bigger than the
-        # pipe buffer would block the write forever without a bound on it.
-        process.send_signal(signal.SIGSTOP)
+        started = time.monotonic()
+        reply = await worker.run(
+            "import os, signal\nos.kill(os.getpid(), signal.SIGSTOP)"
+        )
+        assert isinstance(reply, Failure)
+        assert "crashed" in reply.message
+        # Killed when it stopped, rather than when the call timed out.
+        assert time.monotonic() - started < 10
+        assert process.returncode == -signal.SIGKILL
+        reply = await worker.run("1 + 1")
+        assert isinstance(reply, Result)
+        assert reply.value == 2
+
+
+async def test_a_worker_that_stops_reading_fails_the_call_instead_of_hanging(tmp_path):
+    # A worker that says it is ready and then never reads stdin. Writing it a
+    # call bigger than the pipe buffer blocks until the write's time limit.
+    script = tmp_path / "worker.py"
+    script.write_text(
+        "import sys, time\n"
+        f"sys.stdout.buffer.write({encode_message(Ready())!r})\n"
+        "sys.stdout.flush()\n"
+        "time.sleep(60)\n"
+    )
+    backend = LocalBackend(worker_script=script, terminate_grace=0.1)
+    async with make_worker(backend=backend, call_timeout=0.5) as worker:
         reply = await worker.run("x = " + "1" * (1024 * 1024))
         assert isinstance(reply, Failure)
         assert "stopped reading" in reply.message