From c489fc1defbd3330a47107ec7071e3bef2c717fa Mon Sep 17 00:00:00 2001 From: weihaoxuan <641627652@qq.com> Date: Tue, 8 Sep 2026 16:37:15 +0800 Subject: [PATCH] fix(code-run): terminate descendant processes --- frontends/tests/test_code_run_process_tree.py | 72 +++++++++++++++++++ ga.py | 66 +++++++++++++++-- 2 files changed, 133 insertions(+), 5 deletions(-) create mode 100644 frontends/tests/test_code_run_process_tree.py diff --git a/frontends/tests/test_code_run_process_tree.py b/frontends/tests/test_code_run_process_tree.py new file mode 100644 index 000000000..30235be87 --- /dev/null +++ b/frontends/tests/test_code_run_process_tree.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import importlib.util +import sys +import threading +import time +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(ROOT)) + +spec = importlib.util.spec_from_file_location("ga_code_run_under_test", ROOT / "ga.py") +ga = importlib.util.module_from_spec(spec) +assert spec.loader is not None +previous_agent_loop = sys.modules.pop("agent_loop", None) +try: + spec.loader.exec_module(ga) +finally: + if previous_agent_loop is not None: sys.modules["agent_loop"] = previous_agent_loop + + +def _run_with_child(tmp_path, *, timeout, stop_signal, ignore_term=False): + started = tmp_path / "child-started" + canary = tmp_path / "orphan-canary" + child = ( + "import pathlib,signal,time; " + + ("signal.signal(signal.SIGTERM, signal.SIG_IGN); " if ignore_term else "") + + f"pathlib.Path({str(started)!r}).write_text('started'); " + "time.sleep(2); " + f"pathlib.Path({str(canary)!r}).write_text('orphaned')" + ) + parent = ( + "import subprocess,sys,time; " + f"subprocess.Popen([sys.executable, '-c', {child!r}]); " + "time.sleep(30)" + ) + result = None + runner = ga.code_run(parent, "python", timeout, str(tmp_path), str(tmp_path), stop_signal) + try: + while True: next(runner) + except StopIteration as exc: result = exc.value + return result, started, canary + + +def test_manual_stop_terminates_descendants(tmp_path): + stop_signal = [] + outcome = {} + + def run(): outcome["result"], outcome["started"], outcome["canary"] = _run_with_child( + tmp_path, timeout=10, stop_signal=stop_signal) + + thread = threading.Thread(target=run) + thread.start() + deadline = time.time() + 5 + while not (tmp_path / "child-started").exists() and time.time() < deadline: time.sleep(0.02) + assert (tmp_path / "child-started").exists() + stop_signal.append(1) + thread.join(timeout=5) + assert not thread.is_alive() + time.sleep(2.2) + assert not outcome["canary"].exists() + assert "[Stopped]" in outcome["result"]["stdout"] + + +def test_timeout_terminates_descendants(tmp_path): + result, started, canary = _run_with_child( + tmp_path, timeout=0.1, stop_signal=[], ignore_term=True) + assert started.exists() + time.sleep(2.2) + assert not canary.exists() + assert "[Timeout Error]" in result["stdout"] diff --git a/ga.py b/ga.py index e1ee909f2..e747c67f1 100644 --- a/ga.py +++ b/ga.py @@ -1,4 +1,4 @@ -import sys, os, re, json, time, threading, importlib, webbrowser +import sys, os, re, json, time, threading, importlib, webbrowser, signal from datetime import datetime from pathlib import Path import tempfile, traceback, subprocess, itertools, collections, difflib, shutil @@ -13,6 +13,57 @@ def safe_print(*args, **kwargs): try: print(*args, **kwargs) except: pass +def _windows_job(process): + if os.name != 'nt': return None + try: + import ctypes + from ctypes import wintypes + k32 = ctypes.WinDLL('kernel32', use_last_error=True) + k32.CreateJobObjectW.restype = wintypes.HANDLE + k32.AssignProcessToJobObject.argtypes = [wintypes.HANDLE, wintypes.HANDLE] + k32.AssignProcessToJobObject.restype = wintypes.BOOL + k32.TerminateJobObject.argtypes = [wintypes.HANDLE, wintypes.UINT] + k32.TerminateJobObject.restype = wintypes.BOOL + k32.CloseHandle.argtypes = [wintypes.HANDLE] + handle = k32.CreateJobObjectW(None, None) + if not handle: return None + if not k32.AssignProcessToJobObject(handle, process._handle): + k32.CloseHandle(handle); return None + return k32, handle + except OSError: + return None + +def _close_windows_job(job, kill=False): + if not job: return False + k32, handle = job + stopped = bool(k32.TerminateJobObject(handle, 1)) if kill else True + k32.CloseHandle(handle) + return stopped + +def _stop_process_tree(process, job=None): + if os.name == 'nt': + stopped = _close_windows_job(job, kill=True) if job else False + if not stopped and process.poll() is None: + subprocess.run(['taskkill', '/PID', str(process.pid), '/T', '/F'], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=0x08000000) + else: + try: os.killpg(process.pid, signal.SIGTERM) + except (ProcessLookupError, PermissionError): pass + deadline = time.monotonic() + 1 + while time.monotonic() < deadline: + try: os.killpg(process.pid, 0) + except ProcessLookupError: break + except PermissionError: pass + time.sleep(0.05) + else: + try: os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): pass + if process.poll() is None: + try: process.kill() + except ProcessLookupError: pass + try: process.wait(timeout=1) + except subprocess.TimeoutExpired: pass + def code_run(code, code_type="python", timeout=60, cwd=None, code_cwd=None, stop_signal=None, maxlen=10000, myprint=safe_print): """代码执行器 python: 运行复杂的 .py 脚本（文件模式） @@ -59,8 +110,10 @@ def stream_reader(proc, logs): cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0, cwd=cwd, startupinfo=startupinfo, - creationflags=0x08000000 if os.name == 'nt' else 0 + creationflags=0x08000000 if os.name == 'nt' else 0, + start_new_session=os.name != 'nt' ) + process_job = _windows_job(process) start_t = time.time() t = threading.Thread(target=stream_reader, args=(process, full_stdout), daemon=True) t.start() @@ -68,8 +121,9 @@ def stream_reader(proc, logs): while t.is_alive(): istimeout = time.time() - start_t > timeout if istimeout or stop_signal: - process.kill() - myprint("[Debug] Process killed due to timeout or stop signal.") + _stop_process_tree(process, process_job) + process_job = None + myprint("[Debug] Process tree stopped due to timeout or stop signal.") if istimeout: full_stdout.append("\n[Timeout Error] 超时强制终止") else: full_stdout.append("\n[Stopped] 用户强制终止") break @@ -92,9 +146,11 @@ def stream_reader(proc, logs): "exit_code": exit_code } except Exception as e: - if 'process' in locals(): process.kill() + if 'process' in locals(): + _stop_process_tree(process, locals().get('process_job')) return {"status": "error", "msg": str(e)} finally: + if 'process_job' in locals() and process_job: _close_windows_job(process_job) if code_type == "python" and tmp_path and os.path.exists(tmp_path): os.remove(tmp_path)