fix isolated_runner: TimeoutExpired has no .pid (e-261003004801) — Popen start_new_session + killpg(proc.pid), group+children reaped; scheduler None-guard in r2 double-close guard (mocked es in tests); regression test test_timeout_kill

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/<pid>/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")
+            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")

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:

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)