0.8.4: a stopped step stops everything it started

When a step was stopped (its timeout ran out, the run was canceled, or
the runner was shutting down) the runner killed only the shell that
wraps the step's command. Whatever ran under that shell carried on with
nobody watching: on 2026-11-11 the org-root dataset-scout seed step hit
its 240 minute timeout and its claude process kept working against the
server for another 40 minutes.

Each step now runs in its own session and process group. On a stop the
runner notes every process under the step (one that moved to a session
of its own is still found, by parentage), sends them SIGTERM so a
program that manages children of its own can stop them, and kills
whatever is still running 5 seconds later. Windows is unchanged.

The old cancel test only looked at the shell's own return code, which
is why it never saw this. Four tests now cover a grandchild, a
grandchild in its own session, a process that ignores SIGTERM, and that
a step never shares the runner's process group.

Author: ninthreezy

CHANGELOG.md

@@ -3,6 +3,30 @@
 Notable changes to the `datatorch` python package (CLI, SDK and the runner
 daemon). Versions are the tags on this repository and the releases on PyPI.
 
+## 0.8.4 (2026-11-12)
+
+### Fixed
+
+- Runner: a step that is stopped (its timeout ran out, the run was
+  canceled, or the runner is shutting down) now stops everything it
+  started, not only the shell that wraps its command. Before, the runner
+  killed that one shell and whatever ran under it carried on with nobody
+  watching: on 2026-11-11 the org-root `dataset-scout` seed step hit its
+  240 minute timeout and its `claude` process kept working against the
+  server for another 40 minutes. Each step now runs in its own session
+  and process group. On a stop the runner notes every process under the
+  step (one that moved to a session of its own is still found, by
+  parentage), sends them SIGTERM so a program that manages children of
+  its own can stop them, and kills whatever is still running 5 seconds
+  later (`TERMINATE_GRACE_SECONDS`). Windows is unchanged: the direct
+  child is terminated as before.
+
+### Changed
+
+- Runner: because a step has its own session, Ctrl-C in the terminal the
+  runner was started from reaches the runner alone. The runner then stops
+  its running steps itself, through the path above.
+
 ## 0.8.3 (2026-11-09)
 
 ### Fixed

datatorch/runner/pipelines/runner/runner.py

@@ -1,8 +1,12 @@
 import os
 import json
+import time
+import signal
 import typing
 import asyncio
 
+import psutil
+
 from typing import Any, Awaitable, Dict
 from ..template import Variables
 
@@ -19,6 +23,51 @@ class ProcessCodeError(Exception):
 # output value (e.g. an http-request body), not ordinary log lines.
 STDOUT_LINE_LIMIT = 16 * 1024 * 1024
 
+# How long a stopped step's processes get to exit by themselves (SIGTERM)
+# before they are killed. The server has already closed the step by then,
+# so nothing waits on this except the processes' own cleanup.
+TERMINATE_GRACE_SECONDS = 5.0
+
+# Windows has no process groups in this sense and no SIGKILL.
+_POSIX = os.name == "posix"
+_SIGKILL = getattr(signal, "SIGKILL", signal.SIGTERM)
+
+
+def _descendants(pid: int) -> list:
+    """Every process under `pid` right now, as psutil handles (which
+    refuse to signal a pid that has since been reused)."""
+    try:
+        return psutil.Process(pid).children(recursive=True)
+    except psutil.Error:
+        return []
+
+
+def _alive(proc) -> bool:
+    try:
+        return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE
+    except psutil.Error:
+        return False
+
+
+def _signal_tree(process, tree: list, sig: int) -> None:
+    """Send `sig` to the step's process group and to every noted
+    descendant, wherever it now lives. A process already gone is fine."""
+    if process.returncode is None:
+        try:
+            if _POSIX:
+                # run_cmd gives each step its own session, so its group id
+                # is the shell's pid and never the runner's own group.
+                os.killpg(process.pid, sig)
+            else:
+                process.send_signal(sig)
+        except (ProcessLookupError, PermissionError, OSError):
+            pass
+    for proc in tree:
+        try:
+            proc.send_signal(sig)
+        except (psutil.Error, OSError):
+            pass
+
 
 class Runner(object):
     def __init__(self, config: dict, action: "Action"):
@@ -44,8 +93,8 @@ async def monitor_cmd(self, command: str, env: dict = None):
 
         Streams stdout live (rather than waiting then draining) so that if
         the step is canceled or times out server-side — the asyncio task
-        running it receives ``CancelledError`` — the child process is killed
-        promptly instead of being orphaned.
+        running it receives ``CancelledError`` — the child process and
+        everything it started are stopped instead of being orphaned.
         """
         process = await self.run_cmd(command, wait=False, env=env)
 
@@ -69,11 +118,36 @@ async def monitor_cmd(self, command: str, env: dict = None):
             )
 
     async def _terminate(self, process):
-        """Kill a still-running child process (best effort)."""
+        """Stop a still-running step: the whole process tree, not only the
+        shell that wraps the command (best effort).
+
+        `run_cmd` starts the command through a shell, so `process` is that
+        shell. Killing it alone left everything under it running: on
+        2026-11-11 a timed-out seed step's `claude` process worked on for
+        40 minutes after the server had closed the step. So: note every
+        descendant first (a process that moved to its own group is found by
+        parentage, and once its parent dies that link is gone), ask the
+        tree to stop with SIGTERM so a process that manages children of its
+        own can stop them, then kill whatever is left after
+        `TERMINATE_GRACE_SECONDS`.
+        """
         if process.returncode is not None:
             return
+
+        tree = _descendants(process.pid)
+        _signal_tree(process, tree, signal.SIGTERM)
+
+        deadline = time.monotonic() + TERMINATE_GRACE_SECONDS
+        while time.monotonic() < deadline:
+            if process.returncode is not None and not any(
+                _alive(proc) for proc in tree
+            ):
+                break
+            await asyncio.sleep(0.1)
+        else:
+            _signal_tree(process, tree, _SIGKILL)
+
         try:
-            process.kill()
             await process.wait()
         except ProcessLookupError:
             pass
@@ -157,6 +231,10 @@ async def run_cmd(self, command: str, wait: bool = True, env: dict = None):
             # page body — asyncio's 64KB default readline limit kills the
             # stream with LimitOverrunError on anything larger.
             limit=STDOUT_LINE_LIMIT,
+            # Its own session and process group, so a stop can signal the
+            # step's whole tree at once without touching the runner (see
+            # _terminate). Ignored on Windows.
+            start_new_session=True,
         )
         if wait:
             await process.wait()

setup.py

@@ -33,7 +33,7 @@
 
 setup(
     name="datatorch",
-    version="0.8.3",
+    version="0.8.4",
     description="A CLI and library for interacting with DataTorch.",
     author="DataTorch",
     author_email="support@datatorch.io",

test/runner/pipelines/test_runner_cancel.py

@@ -6,9 +6,15 @@
 """
 
 import asyncio
+import os
+import shlex
+import sys
+import time
 
+import psutil
 import pytest
 
+from datatorch.runner.pipelines.runner import runner as runner_module
 from datatorch.runner.pipelines.runner.runner import Runner
 
 
@@ -47,3 +53,110 @@ async def capturing(command, wait=True, env=None):
     # A short wall-clock timeout guards against a regression where cancel
     # hangs waiting for the full `sleep 30`.
     asyncio.run(asyncio.wait_for(inner(), timeout=10))
+
+
+# --- The whole tree stops, not only the shell (0.8.4) -----------------------
+#
+# On 2026-11-11 a seed step hit its timeout, the runner killed the shell that
+# wrapped the action, and the `claude` process under it kept working against
+# the server for another 40 minutes. The test above could not see that: it
+# only looked at the shell's own return code.
+
+
+def _gone(pid: int) -> bool:
+    try:
+        proc = psutil.Process(pid)
+        return proc.status() == psutil.STATUS_ZOMBIE
+    except psutil.NoSuchProcess:
+        return True
+
+
+def _cancel_and_collect(command: str, pid_file: str, expected: int = 1):
+    """Run `command` as a step, wait until it has written `expected` pids
+    to `pid_file`, cancel the step, and return those pids."""
+
+    async def inner():
+        runner = Runner({}, _FakeAction())
+        task = asyncio.ensure_future(runner.monitor_cmd(command))
+
+        pids = []
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline and len(pids) < expected:
+            await asyncio.sleep(0.05)
+            if os.path.exists(pid_file):
+                with open(pid_file) as handle:
+                    pids = [int(part) for part in handle.read().split()]
+        assert len(pids) == expected, "the step never started its children"
+        assert all(psutil.pid_exists(pid) for pid in pids)
+
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        return pids
+
+    return asyncio.run(asyncio.wait_for(inner(), timeout=20))
+
+
+@pytest.mark.skipif(os.name != "posix", reason="process groups are POSIX")
+def test_cancel_stops_a_grandchild(tmp_path):
+    """The shape of a python action: the shell starts a program, and that
+    program starts another. Both must be gone after a cancel."""
+    pid_file = str(tmp_path / "pids")
+    command = f"(sleep 60 & echo $! > {shlex.quote(pid_file)}; wait)"
+
+    (grandchild,) = _cancel_and_collect(command, pid_file)
+
+    assert _gone(grandchild)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="sessions are POSIX")
+def test_cancel_stops_a_grandchild_in_its_own_session(tmp_path):
+    """A program that puts its children in their own session (an agent's
+    shell tool does) takes them out of the step's process group. They are
+    still found, by parentage, and stopped."""
+    pid_file = str(tmp_path / "pids")
+    script = (
+        "import subprocess, sys;"
+        "p = subprocess.Popen(['sleep', '60'], start_new_session=True);"
+        "open(sys.argv[1], 'w').write(str(p.pid));"
+        "p.wait()"
+    )
+    command = " ".join(
+        shlex.quote(part) for part in (sys.executable, "-c", script, pid_file)
+    )
+
+    (grandchild,) = _cancel_and_collect(command, pid_file)
+
+    assert _gone(grandchild)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="signals are POSIX")
+def test_cancel_kills_what_ignores_the_polite_stop(tmp_path, monkeypatch):
+    """A process that ignores SIGTERM is killed once the grace runs out."""
+    monkeypatch.setattr(runner_module, "TERMINATE_GRACE_SECONDS", 0.5)
+    pid_file = str(tmp_path / "pids")
+    command = f"trap '' TERM; (sleep 60 & echo $! > {shlex.quote(pid_file)}; wait)"
+
+    started = time.monotonic()
+    (grandchild,) = _cancel_and_collect(command, pid_file)
+
+    assert _gone(grandchild)
+    assert time.monotonic() - started < 10
+
+
+@pytest.mark.skipif(os.name != "posix", reason="process groups are POSIX")
+def test_a_step_runs_outside_the_runners_process_group():
+    """The stop signals a whole group, so that group must never be the
+    runner's own."""
+
+    async def inner():
+        runner = Runner({}, _FakeAction())
+        process = await runner.run_cmd("sleep 60", wait=False)
+        try:
+            assert os.getpgid(process.pid) == process.pid
+            assert os.getpgid(process.pid) != os.getpgrp()
+        finally:
+            await runner._terminate(process)
+        assert process.returncode is not None
+
+    asyncio.run(asyncio.wait_for(inner(), timeout=10))