fix(sandbox): use saved process group id on timeout (#7353)

* fix(sandbox): retain process group id for timeout cleanup

* test(sandbox): cover exited leader during group cleanup

* fix(sandbox): capture process group id before drain threads

* style(sandbox): separate timeout regression test

* test(sandbox): cover missing process-group fallback

* style(tests): format sandbox timeout regressions

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/sandbox/local/local_sandbox.py

@@ -1021,6 +1021,9 @@ class LocalSandbox(Sandbox):
                 start_new_session=True,
                 env=env,
             )
+            # With start_new_session=True, the new process is also the
+            # process-group leader, so its PID is the stable group identifier.
+            process_group_id = process.pid
         except Exception:
             for fd in (stdout_read_fd, stdout_write_fd, stderr_read_fd, stderr_write_fd):
                 try:
@@ -1039,17 +1042,12 @@ class LocalSandbox(Sandbox):
 
         stdout_capture, stdout_thread = LocalSandbox._start_pipe_drain(stdout_read_fd, "deerflow-bash-stdout-drain")
         stderr_capture, stderr_thread = LocalSandbox._start_pipe_drain(stderr_read_fd, "deerflow-bash-stderr-drain")
-        try:
-            process_group_id = os.getpgid(process.pid)
-        except OSError:
-            process_group_id = None
-
         try:
             try:
                 process.wait(timeout=timeout)
             except subprocess.TimeoutExpired:
                 timed_out = True
-                LocalSandbox._terminate_process_group(process)
+                LocalSandbox._terminate_process_group(process, process_group_id)
             returncode = process.returncode if process.returncode is not None else 0
         finally:
             join_timeout = 10 if timed_out or not LocalSandbox._process_group_exists(process_group_id) else _PIPE_DRAIN_JOIN_TIMEOUT_SECONDS
@@ -1063,14 +1061,14 @@ class LocalSandbox(Sandbox):
         return stdout, stderr, returncode, timed_out
 
     @staticmethod
-    def _terminate_process_group(process: subprocess.Popen) -> None:
+    def _terminate_process_group(process: subprocess.Popen, process_group_id: int) -> None:
         """Kill the command's whole process group, then reap it.
 
         Falls back to killing just the direct child if the group is already
         gone (e.g. the command exited between the timeout and this call).
         """
         try:
-            os.killpg(os.getpgid(process.pid), signal.SIGKILL)
+            os.killpg(process_group_id, signal.SIGKILL)
         except (ProcessLookupError, PermissionError, OSError):
             # The process group is already gone (the command exited in the race
             # between the timeout and this call); fall back to killing just the

backend/tests/test_local_sandbox_command_timeout.py

@@ -11,6 +11,8 @@ semantics, while the shared tests pin the user-facing timeout notice.
 
 import os
 import shlex
+import signal
+import subprocess
 import sys
 import time
 from pathlib import Path
@@ -221,3 +223,79 @@ def test_bash_tool_description_guides_safe_cross_platform_local_environment_prob
     assert "only when the active sandbox policy permits it" in description
     assert "do not repeat the rejected command" in description
     assert "command-only probes" in description
+
+
+@posix_only
+def test_timeout_uses_saved_process_group_id_if_leader_exits(monkeypatch):
+    """Timeout cleanup must target the saved group even if the leader has exited."""
+    calls = []
+
+    class ExitedLeaderProcess:
+        pid = 4242
+
+        def __init__(self):
+            self.returncode = None
+            self.wait_calls = 0
+
+        def wait(self, timeout=None):
+            self.wait_calls += 1
+            if self.wait_calls == 1:
+                # The command crossed the deadline as its leader exited; its
+                # descendants may still hold the captured pipes open.
+                self.returncode = 0
+                raise subprocess.TimeoutExpired("fake-command", timeout)
+            return self.returncode
+
+        def kill(self):
+            pytest.fail("fallback child kill should not replace process-group cleanup")
+
+    process = ExitedLeaderProcess()
+    monkeypatch.setattr(local_sandbox.subprocess, "Popen", lambda *args, **kwargs: process)
+    monkeypatch.setattr(
+        local_sandbox.os,
+        "getpgid",
+        lambda pid: pytest.fail("cleanup must not re-query an exited process leader"),
+    )
+    monkeypatch.setattr(
+        local_sandbox.os,
+        "killpg",
+        lambda process_group_id, sig: calls.append((process_group_id, sig)),
+    )
+
+    _, _, returncode, timed_out = LocalSandbox._run_posix_command(["fake-command"], timeout=0.01)
+
+    assert timed_out is True
+    assert returncode == 0
+    assert calls == [(process.pid, signal.SIGKILL)]
+
+
+@posix_only
+def test_process_group_lookup_failure_falls_back_to_direct_child_kill(monkeypatch):
+    """A vanished process group is tolerated by killing/reaping the direct child."""
+    calls = []
+
+    class FakeProcess:
+        pid = 12345
+
+        def kill(self):
+            calls.append(("kill-child",))
+
+        def wait(self, timeout=None):
+            calls.append(("wait", timeout))
+
+    process = FakeProcess()
+
+    def missing_process_group(process_group_id, sig):
+        calls.append(("killpg", process_group_id, sig))
+        raise ProcessLookupError
+
+    monkeypatch.setattr(local_sandbox.os, "killpg", missing_process_group)
+
+    # ESRCH from killpg must not escape; the direct-child fallback still runs.
+    LocalSandbox._terminate_process_group(process, process.pid)
+
+    assert calls == [
+        ("killpg", process.pid, signal.SIGKILL),
+        ("kill-child",),
+        ("wait", 10),
+    ]