fix(sandbox): terminate POSIX shell children on timeout

ag2/tools/sandbox/local.py

@@ -6,6 +6,7 @@
 import atexit
 import os
 import shutil
+import signal
 import subprocess
 import tempfile
 from contextlib import suppress
@@ -39,31 +40,40 @@ def _run_subprocess(
     timeout: float,
     max_output: int,
 ) -> ExecResult:
-    """Run *argv* under :func:`subprocess.run` with the project's conventions."""
+    """Run *argv*, terminating its POSIX process group on timeout."""
     merged_env = {**os.environ, **env} if env is not None else None
 
     try:
-        result = subprocess.run(
+        with subprocess.Popen(
             argv,
             cwd=cwd,
-            capture_output=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
             text=True,
-            timeout=timeout,
             env=merged_env,
-        )
+            start_new_session=os.name == "posix",
+        ) as process:
+            try:
+                stdout, stderr = process.communicate(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                if os.name == "posix":
+                    with suppress(ProcessLookupError):
+                        os.killpg(process.pid, signal.SIGKILL)
+                else:
+                    process.kill()
+                process.wait()
+                return ExecResult(output=f"Command timed out after {timeout}s", exit_code=124)
+            except BaseException:
+                process.kill()
+                raise
     except FileNotFoundError as e:
         return ExecResult(output=f"Command not found: {e}", exit_code=127)
-    except subprocess.TimeoutExpired:
-        return ExecResult(
-            output=f"Command timed out after {timeout}s",
-            exit_code=124,
-        )
 
-    output = (result.stdout + result.stderr).strip()
+    output = (stdout + stderr).strip()
     if (total := len(output)) > max_output:
         output = output[:max_output]
         output += f"\n[truncated: showing first {max_output} of {total} chars]"
-    return ExecResult(output=output, exit_code=result.returncode or 0)
+    return ExecResult(output=output, exit_code=process.returncode or 0)
 
 
 class LocalSandbox(SandboxBase):
@@ -80,7 +90,8 @@ class LocalSandbox(SandboxBase):
                  backstop). Defaults to ``True`` when ``path=None``
                  (auto temp dir) and ``False`` otherwise.
         timeout: Default per-call timeout in seconds. Must be ``> 0``.
-                 Overridable per :meth:`exec` call.
+                 Overridable per :meth:`exec` call. On POSIX, timeout
+                 terminates the command's process group, including shell children.
         max_output: Maximum number of characters in :attr:`ExecResult.output`.
                     Excess is truncated with a trailing notice.
         env_vars: Default environment variables merged into every command

test/tools/sandbox/test_local_sandbox.py

@@ -2,6 +2,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import asyncio
+import os
+import shlex
+import sys
 from pathlib import Path, PurePosixPath
 
 import pytest
@@ -54,6 +58,27 @@ async def test_per_call_timeout_overrides_default(self) -> None:
         result = await sandbox.exec(["sleep", "5"], timeout=0.1)
         assert result.exit_code == 124
 
+    @pytest.mark.skipif(os.name != "posix", reason="POSIX shell process groups")
+    @pytest.mark.parametrize("ending", ["; echo done", " &"])
+    async def test_timeout_stops_shell_children(self, tmp_path: Path, ending: str) -> None:
+        program = (
+            "from pathlib import Path; import time; "
+            "Path('started').touch(); "
+            "deadline = time.monotonic() + 3; "
+            "\nwhile not Path('release').exists() and time.monotonic() < deadline: time.sleep(0.01)"
+            "\nif Path('release').exists(): Path('completed').touch()"
+        )
+        command = f"{shlex.quote(sys.executable)} -c {shlex.quote(program)}{ending}"
+        async with LocalSandbox(tmp_path, timeout=0.5) as sandbox:
+            result = await sandbox.exec(["sh", "-c", command])
+            assert result.exit_code == 124
+            assert (tmp_path / "started").exists()
+
+            (tmp_path / "release").touch()
+            await asyncio.sleep(0.5)
+
+        assert not (tmp_path / "completed").exists()
+
     async def test_output_is_truncated_when_too_long(self) -> None:
         sandbox = LocalSandbox(max_output=10)
         long = "x" * 200