fix: terminate detached descendants on bbdev cancellation

api/unit_tests/test_process_cleanup.py

@@ -0,0 +1,72 @@
+import os
+import shlex
+import signal
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from utils.process import _exists, terminate_group
+from utils.process_registry import cancel_process, register_process, unregister_process
+from utils.stream_run import stream_run
+
+
+class ProcessCleanupTest(unittest.TestCase):
+    def test_nested_session_exits_on_shutdown_and_cancel(self):
+        for cancel in (False, True):
+            with self.subTest(cancel=cancel):
+                parent = subprocess.Popen(
+                    [
+                        sys.executable,
+                        "-c",
+                        "import subprocess, sys, time; "
+                        "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'], start_new_session=True); "
+                        "print(child.pid, flush=True); time.sleep(300)",
+                    ],
+                    stdout=subprocess.PIPE,
+                    text=True,
+                    start_new_session=True,
+                )
+                child_pid = int(parent.stdout.readline())
+                try:
+                    if cancel:
+                        register_process("nested-session-test", parent.pid)
+                        self.assertTrue(cancel_process("nested-session-test"))
+                        parent.wait(timeout=2)
+                    else:
+                        terminate_group(parent, term_timeout=1)
+                    self.assertFalse(_exists(child_pid))
+                finally:
+                    unregister_process("nested-session-test")
+                    if _exists(child_pid):
+                        os.kill(child_pid, signal.SIGKILL)
+                    if parent.poll() is None:
+                        parent.kill()
+                    parent.wait()
+                    parent.stdout.close()
+
+    def test_timeout_exits_nested_session(self):
+        with tempfile.TemporaryDirectory() as directory:
+            child_pid_file = Path(directory) / "child.pid"
+            command = shlex.join(
+                [
+                    sys.executable,
+                    "-c",
+                    "import subprocess, sys, time; "
+                    "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'], start_new_session=True); "
+                    "open(sys.argv[1], 'w').write(str(child.pid)); time.sleep(300)",
+                    str(child_pid_file),
+                ]
+            )
+            try:
+                result = stream_run(command, timeout=1)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertFalse(_exists(int(child_pid_file.read_text())))
+            finally:
+                if child_pid_file.exists():
+                    child_pid = int(child_pid_file.read_text())
+                    if _exists(child_pid):
+                        os.kill(child_pid, signal.SIGKILL)

api/utils/process.py

@@ -113,27 +113,8 @@ def kill_tree(pid: int, term_timeout: float = 5.0):
 def terminate_group(proc: subprocess.Popen, term_timeout: float = 5.0):
   if proc.poll() is not None:
     return
-
-  try:
-    os.killpg(proc.pid, signal.SIGTERM)
-  except (ProcessLookupError, PermissionError, OSError):
-    kill_tree(proc.pid, term_timeout=term_timeout)
-
-  try:
-    proc.wait(timeout=term_timeout)
-    return
-  except subprocess.TimeoutExpired:
-    pass
-
-  try:
-    os.killpg(proc.pid, signal.SIGKILL)
-  except (ProcessLookupError, PermissionError, OSError):
-    kill_tree(proc.pid, term_timeout=1.0)
-
-  try:
-    proc.wait(timeout=1)
-  except subprocess.TimeoutExpired:
-    pass
+  kill_tree(proc.pid, term_timeout=term_timeout)
+  proc.wait(timeout=1)
 
 
 def _is_workspace_server(pid: int, workflow_dir: str, worker_url: str | None = None) -> bool:

api/utils/process_registry.py

@@ -6,6 +6,8 @@
 from contextvars import ContextVar
 from typing import Dict
 
+from utils.process import kill_tree
+
 
 _lock = threading.Lock()
 _processes: Dict[str, int] = {}
@@ -38,10 +40,7 @@ def cancel_process(scope: str) -> bool:
         _cancelled.add(scope)
     if pid is None:
         return False
-    try:
-        os.killpg(pid, signal.SIGTERM)
-    except ProcessLookupError:
-        return False
+    kill_tree(pid)
     return True
 
 

api/utils/stream_run.py

@@ -3,6 +3,7 @@
 import threading
 from typing import Optional, List, Callable
 
+from utils.process import kill_tree
 from utils.process_registry import current_task_scope, register_process, unregister_process
 
 
@@ -133,8 +134,7 @@ def emit():
         # Wait for process to finish (with timeout)
         process.wait(timeout=timeout)
     except subprocess.TimeoutExpired:
-        # Kill process on timeout
-        process.kill()
+        kill_tree(process.pid)
         process.wait()
 
     # Wait for threads to finish