From 03d3dffeb8be4c8eecf86d33d29afa6f1e1c76ba Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Sat, 3 Oct 2026 01:17:34 +0000 Subject: [PATCH] =?UTF-8?q?fix=20isolated=5Frunner:=20TimeoutExpired=20has?= =?UTF-8?q?=20no=20.pid=20(e-261003004801)=20=E2=80=94=20Popen=20start=5Fn?= =?UTF-8?q?ew=5Fsession=20+=20killpg(proc.pid),=20group+children=20reaped;?= =?UTF-8?q?=20scheduler=20None-guard=20in=20r2=20double-close=20guard=20(m?= =?UTF-8?q?ocked=20es=20in=20tests);=20regression=20test=20test=5Ftimeout?= =?UTF-8?q?=5Fkill?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- harness/isolated_runner.py | 97 +++++++++++++++++++++----------------- harness/scheduler.py | 5 +- tests/test_timeout_kill.py | 79 +++++++++++++++++++++++++++++++ 3 files changed, 135 insertions(+), 46 deletions(-) create mode 100644 tests/test_timeout_kill.py diff --git a/harness/isolated_runner.py b/harness/isolated_runner.py index 870ebc7..ddea8ae 100644 --- a/harness/isolated_runner.py +++ b/harness/isolated_runner.py @@ -50,61 +50,70 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim # просто поглощает такты (TickAbsorbed), это норма архитектуры timeout = getattr(level_cfg, "tick_timeout", 300) timeout = max(timeout, int(level_cfg.every * 0.8)) + # start_new_session: воркер становится лидером СВОЕЙ процесс-группы — + # добивание на таймауте достаёт всю группу LLM-потомков по proc.pid, + # без чтения /proc//status. + proc = subprocess.Popen( + [_VENV_PY, _WORKER], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=_HARNESS_DIR, + env=env, + start_new_session=True, + ) try: - proc = subprocess.run( - [_VENV_PY, _WORKER], - input=json.dumps(payload).encode(), - capture_output=True, - timeout=timeout, - cwd=_HARNESS_DIR, - env=env, - ) - except subprocess.TimeoutExpired as e: - # воркер убит по таймауту. subprocess.run убивает только основной - # процесс (Popen.kill через os.kill без группы) — сабпроцесс _llm_worker - # успевал сам заспавнить LLM-подпроцессы, которые оставались ЖИВЫМИ и - # продолжали писать в стор после SchedulerTimeout (BudgetSpent/ToolCalled - # с завершённым trace — race, burn 9router, t-261002090242-cf7cf9). - # Добиваем всю процесс-группу и ЖДЁМ фактической смерти. - pgid = None + stdout, stderr = proc.communicate( + input=json.dumps(payload).encode(), timeout=timeout) + except subprocess.TimeoutExpired: + # воркер не уложился в таймаут. subprocess.communicate() бросает + # TimeoutExpired БЕЗ .pid — subprocess.run не отдавал Popen, и + # аварийная зачистка падала с AttributeError ('TimeoutExpired' + # object has no attribute 'pid', e-261003004801) ДО добивания + # группы: LLM-потомки выживали и писали в стор после смерти тика + # (race, burn роутера, t-261002090242). С Popen pid есть всегда. + pgid = proc.pid try: - with open(f"/proc/{e.pid}/status") as f: - for line in f: - if line.startswith("Tgid:"): - pgid = line.split()[1] - break - except OSError: - pass - try: - if e.pid > 1: - os.killpg(pgid or e.pid, signal.SIGKILL) + if pgid > 1: + os.killpg(pgid, signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): pass for _ in range(10): # до 5с на смерть группы, иначе — SIGKILL ещё раз try: - os.killpg(pgid or e.pid, 0) + os.killpg(pgid, 0) time.sleep(0.5) except (ProcessLookupError, PermissionError) as ex: break else: try: - os.killpg(pgid or e.pid, signal.SIGKILL) + os.killpg(pgid, signal.SIGKILL) + except OSError: + pass + for stream in (proc.stdout, proc.stderr, proc.stdin): + try: + if stream: + stream.close() except OSError: pass - raise - out = None - if os.path.exists(result_path): try: - with open(result_path) as f: - out = json.load(f) - except ValueError: - out = None - try: - os.unlink(result_path) - except OSError: - pass - if out is None: - return None, {"error": f"worker result file missing: stdout={(proc.stdout or b'')[:200]!r} stderr={(proc.stderr or b'').decode(errors='replace')[-300:]}"}, None - if not out.get("ok"): - return None, {"error": (out.get("error") or "unknown")[:500]}, None - return out.get("text", ""), out.get("usage", {}), out.get("trace_id") \ No newline at end of file + proc.wait(timeout=5) # reap: без этого — зомби до смерти родителя + except subprocess.TimeoutExpired: + pass + raise + else: + out = None + if os.path.exists(result_path): + try: + with open(result_path) as f: + out = json.load(f) + except ValueError: + out = None + try: + os.unlink(result_path) + except OSError: + pass + if out is None: + return None, {"error": f"worker result file missing: stdout={(stdout or b'')[:200]!r} stderr={(stderr or b'').decode(errors='replace')[-300:]}"}, None + if not out.get("ok"): + return None, {"error": (out.get("error") or "unknown")[:500]}, None + return out.get("text", ""), out.get("usage", {}), out.get("trace_id") \ No newline at end of file diff --git a/harness/scheduler.py b/harness/scheduler.py index a6d2437..11bbe0b 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -540,9 +540,10 @@ def run_level_tick(level_cfg, seq: int, model: dict): # содержит целевой ответ; повторный append поверх перезаписал бы answer # дайджестом тика и испортил бы проверенный результат (перезапись # «последний-победил» в проекции). Закрываем только ещё-running. - _model = es.run_projections() + _model = es.run_projections() or {} # стаб/None в тестах — не валим + _tasks = _model.get("tasks") or {} _done = {tid for tid in (claimed or []) - if ((_model.get("tasks") or {}).get(tid) or {}).get("status") == "done"} + if (_tasks.get(tid) or {}).get("status") == "done"} _first = True for tid in (claimed or []): if tid in _done: diff --git a/tests/test_timeout_kill.py b/tests/test_timeout_kill.py new file mode 100644 index 0000000..1a3c928 --- /dev/null +++ b/tests/test_timeout_kill.py @@ -0,0 +1,79 @@ +"""Регресс e-261003004801: таймаут LLM-тика не должен валить зачистку. + +Было: subprocess.run + TimeoutExpired без .pid → AttributeError в +изолированном runner'е ДО добивания процесс-группы. LLM-потомки воркера +выживали и писали в стор после смерти тика (race, burn роутера). +Теперь: Popen(start_new_session=True) → killpg(proc.pid); исключение — +честный TimeoutExpired, группа (с потомками) мертва до re-raise. +""" +import os +import subprocess +import sys +import tempfile +import time +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from harness import isolated_runner + + +class TimeoutKillTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="test_timeout_kill_") + self.worker = os.path.join(self.tmp, "fake_worker.py") + # фейковый воркер: спавнит потомка (имитация LLM-сабпроцесса) и спит + with open(self.worker, "w") as f: + f.write( + "import subprocess, time\n" + "child = subprocess.Popen(['sleep', '287'])\n" + "print(child.pid, flush=True)\n" + "time.sleep(300)\n" + ) + self._orig = (isolated_runner._VENV_PY, isolated_runner._WORKER) + isolated_runner._VENV_PY = sys.executable + isolated_runner._WORKER = self.worker + + def tearDown(self): + isolated_runner._VENV_PY, isolated_runner._WORKER = self._orig + + def _cfg(self): + class Cfg: + name = "test" + model = "x" + effort = "low" + mandate = "" + can_shell = False + ctx_tokens = 1000 + every = 300 + return Cfg() + + def test_timeout_kills_group_without_pid_crash(self): + with self.assertRaises(subprocess.TimeoutExpired): + isolated_runner.run_llm_tick(self._cfg(), "ping", timeout=3) + # группа воркера убита: потомок-«LLM» не выжил + time.sleep(0.5) + r = subprocess.run(["pgrep", "-c", "-f", "sleep 287"], + capture_output=True) + # pgrep: exit 1 = совпадений нет → потомок-«LLM» мёртв + self.assertEqual(r.returncode, 1, + f"LLM-потомок пережил таймаут добивания: {r.stdout!r}") + + def test_normal_run_unaffected(self): + # короткий воркер: пишет результат в result_path как настоящий + worker = os.path.join(self.tmp, "ok_worker.py") + with open(worker, "w") as f: + f.write( + "import json, os, sys\n" + "payload = json.loads(sys.stdin.read())\n" + "json.dump({'ok': True, 'text': 'done',\n" + " 'usage': {}, 'trace_id': 't'}, open(payload['result_path'], 'w'))\n" + ) + isolated_runner._WORKER = worker + text, usage, trace = isolated_runner.run_llm_tick(self._cfg(), + "ping", timeout=15) + self.assertEqual(text, "done") + self.assertEqual(trace, "t") + + +if __name__ == "__main__": + unittest.main(verbosity=2) \ No newline at end of file