From c230d04d8317f97f7a4df4539bf85a5f05ed09f0 Mon Sep 17 00:00:00 2001 From: liuhao1024 Date: Sun, 20 Sep 2026 20:26:36 +0800 Subject: [PATCH] fix(tools): reap idle session kernels without a new acquire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The idle sweep lives inside _acquire_kernel, so it only fires when the next kernel request arrives. A host that stays alive but stops executing anything (a pids-exhausted container fail-closing every tool call) never acquires again: idle kernels, their runners and thread pools survive indefinitely, which is how four orphaned runners helped exhaust pids_limit=256 for hours (#117169). A low-frequency daemon reaper — started on the first kernel spawn, using the acquire-path criteria verbatim — retires idle kernels on its own schedule, independent of tool traffic. The same pass sweeps hermes_kernel_* staging dirs untouched for over a week: a live host rmtrees each dir within one idle timeout of the kernel's last use, so a week-old dir belongs to a host that died without cleanup (SIGKILL / container restart). Younger dirs are left alone because a concurrently running host's live kernel may own one; rmtree rejects symlinks rather than following them. --- tests/tools/test_code_kernel.py | 97 +++++++++++++++++++++++++++++++++ tools/code_kernel.py | 64 ++++++++++++++++++++++ 2 files changed, 161 insertions(+) diff --git a/tests/tools/test_code_kernel.py b/tests/tools/test_code_kernel.py index c783ec6633..8a443c5f9a 100644 --- a/tests/tools/test_code_kernel.py +++ b/tests/tools/test_code_kernel.py @@ -505,3 +505,100 @@ class TestPerCellRpcAuthority(unittest.TestCase): _run("y = 2") self.assertIsNot(kernel.cell_authority, first_authority) self.assertFalse(kernel.cell_authority.active) + + +class TestBackgroundIdleReaper(unittest.TestCase): + """#117169: the idle sweep must not depend on the next kernel acquire — a host + that stays alive but wedged (e.g. pids exhaustion fail-closing every tool call) + never acquires again, so a background reaper reapplies the acquire-path criteria + on its own schedule, and staging dirs that outlived a dead host are swept by age.""" + + def _run_as(self, session_key, code, task_id, **kwargs): + from tools.approval_context import reset_current_session_key, set_current_session_key + + token = set_current_session_key(session_key) + try: + return json.loads(execute_code(code, task_id=task_id, **kwargs)) + finally: + reset_current_session_key(token) + + def test_reap_once_sweeps_idle_kernels_without_a_new_acquire(self): + import time as time_module + + from tools.code_kernel import _reap_once + + with _kernel_config(kernel_idle_timeout=1): + self._run_as("conv-a", "x = 41", task_id="turn-1") + stale = next(iter(_KERNELS.values())) + time_module.sleep(1.2) + # No conv-b acquire here: the reaper pass alone must retire the kernel. + _reap_once() + self.assertNotIn(stale.key, _KERNELS) + stale.proc.wait(timeout=10) + self.assertFalse(stale.alive()) + + def test_reap_once_spares_attached_and_fresh_kernels(self): + from tools.code_kernel import _reap_once + + with _kernel_config(kernel_idle_timeout=1): + fresh = self._run_as("conv-fresh", "x = 1", task_id="turn-1") + self.assertEqual(fresh["status"], "success", fresh) + kernel = next(iter(_KERNELS.values())) + kernel.attached += 1 # a cell is mid-flight: reaping must skip it + try: + _reap_once() + self.assertIn(kernel.key, _KERNELS) + self.assertTrue(kernel.alive()) + finally: + kernel.attached -= 1 + + def test_spawn_starts_the_reaper_thread(self): + import threading + + with _kernel_config(): + self._run_as("conv-a", "x = 1", task_id="turn-1") + self.assertTrue(any(t.name == "hermes-kernel-idle-reaper" + for t in threading.enumerate())) + + +class TestStaleStagingDirSweep(unittest.TestCase): + def test_week_old_kernel_dirs_go_and_fresh_ones_stay(self): + import time as time_module + + from tools.code_kernel import _sweep_stale_staging_dirs + + with tempfile.TemporaryDirectory() as tmp: + with patch("tools.code_kernel.tempfile.gettempdir", return_value=tmp): + old = Path(tmp, "hermes_kernel_old") + young = Path(tmp, "hermes_kernel_young") + bystander = Path(tmp, "unrelated_dir") + for path in (old, young, bystander): + path.mkdir() + week_and_a_bit = time_module.time() - 8 * 86400 + os.utime(old, (week_and_a_bit, week_and_a_bit)) + removed = _sweep_stale_staging_dirs() + # Asserted inside the TemporaryDirectory: cleanup would flatten everything. + self.assertEqual(removed, 1) + self.assertFalse(old.exists()) + self.assertTrue(young.exists()) + self.assertTrue(bystander.exists()) + + def test_planted_symlink_is_rejected_not_followed(self): + import time as time_module + + from tools.code_kernel import _sweep_stale_staging_dirs + + with tempfile.TemporaryDirectory() as tmp: + with patch("tools.code_kernel.tempfile.gettempdir", return_value=tmp): + target = Path(tmp, "payload") + target.mkdir() + (target / "keep.txt").write_text("precious", encoding="utf-8") + link = Path(tmp, "hermes_kernel_link") + link.symlink_to(target) + # Both ends look stale, so the sweep reaches rmtree(link) itself. + week_and_a_bit = time_module.time() - 8 * 86400 + os.utime(target, (week_and_a_bit, week_and_a_bit)) + os.utime(link, (week_and_a_bit, week_and_a_bit), follow_symlinks=False) + removed = _sweep_stale_staging_dirs() + self.assertEqual(removed, 0) + self.assertTrue((target / "keep.txt").exists()) diff --git a/tools/code_kernel.py b/tools/code_kernel.py index ee6698df0f..a63b723778 100644 --- a/tools/code_kernel.py +++ b/tools/code_kernel.py @@ -18,11 +18,13 @@ Also hosts what ``tools.code_kernel_remote`` shares: owner resolution, registry, from __future__ import annotations import atexit +import glob import json import logging import os import queue import secrets +import shutil import socket import subprocess import sys @@ -651,6 +653,7 @@ def _spawn(kernel: SessionKernel, *, child_python: str, child_cwd: str, for target, args in ((_rpc_forever, (kernel, max_tool_calls, sandbox_tools)), (_stdout_reader, (kernel,)), (_stderr_reader, (kernel,))): threading.Thread(target=target, args=args, daemon=True).start() + _ensure_background_reaper() def _acquire_kernel(key: Tuple, reset: bool, *, pinned: bool = False) -> Tuple[SessionKernel, bool]: @@ -686,6 +689,67 @@ def _acquire_kernel(key: Tuple, reset: bool, *, pinned: bool = False) -> Tuple[S return kernel, state_reset +# The acquire-path sweep above only fires on the NEXT kernel request. A host that stays +# alive but stops executing anything (a pids-exhausted container whose tool dispatch is +# fail-closed) never acquires again, so idle kernels and their thread pools survive +# indefinitely (#117169). One low-frequency daemon thread reapplies the same criteria on +# its own schedule, independent of tool traffic, and also sweeps staging dirs that +# outlived a host which died without cleanup (SIGKILL / container restart). +_STALE_STAGING_DIR_AGE = 7 * 86400 +_REAPER_INTERVAL_FLOOR, _REAPER_INTERVAL_CEIL = 30.0, 300.0 +_REAPER_GUARD = threading.Lock() + + +def _sweep_stale_staging_dirs(now: Optional[float] = None) -> int: + """Remove ``hermes_kernel_*`` staging dirs untouched for over a week. A live host + rmtrees each dir within one idle timeout of the kernel's last use, so a week-old + dir belongs to a host that died before its cleanup could run; younger dirs are left + alone because a concurrently running host's live kernel may own one. rmtree never + follows symlinks, so a planted link is rejected rather than chased.""" + now = time.time() if now is None else now + removed = 0 + for path in glob.glob(os.path.join(tempfile.gettempdir(), "hermes_kernel_*")): + try: + if now - os.path.getmtime(path) > _STALE_STAGING_DIR_AGE: + # No ignore_errors: a rejected symlink (or a half-removed dir) must not + # count as swept — it stays for the next pass instead. + shutil.rmtree(path) + removed += 1 + except OSError: + continue + return removed + + +def _reap_once() -> None: + """One background pass: the acquire-path idle criteria, then the stale-dir sweep.""" + _, idle_timeout = _lifecycle_limits() + with _REGISTRY.lock: + now = time.monotonic() + expired = [_KERNELS.pop(k) for k in list(_KERNELS) + if _KERNELS[k].attached == 0 and now - _KERNELS[k].last_used > idle_timeout] + for doomed in expired: + doomed.teardown() + _sweep_stale_staging_dirs() + + +def _ensure_background_reaper() -> None: + """Start the reaper once per process (on the first kernel spawn).""" + if _REAPER_GUARD.acquire(blocking=False): + threading.Thread(target=_background_reaper, daemon=True, + name="hermes-kernel-idle-reaper").start() + + +def _background_reaper() -> None: + while True: + _, idle_timeout = _lifecycle_limits() + time.sleep(min(_REAPER_INTERVAL_CEIL, + max(_REAPER_INTERVAL_FLOOR, idle_timeout / 6.0))) + try: + _reap_once() + except Exception: + logger.exception("kernel idle reaper pass failed; retrying next interval") + + def _await_cell(kernel: SessionKernel, timeout: int, is_interrupted) -> Tuple[str, Dict[str, Any]]: """Wait for the cell's reply; returns (host status, payload).""" deadline = time.monotonic() + timeout if timeout else None