diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index 42aa0606218a8..4d8a02f125225 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -1717,6 +1717,71 @@ class TestSystemdCgroupIsolation: assert argv == ["/bin/bash", "-lic", "set +m; echo hello"], argv assert captured["start_new_session"] is True + def test_systemd_post_spawn_failure_never_kills_gateway_process_group( + self, registry, monkeypatch + ): + """The scope wrapper shares the gateway PG, so cleanup must not killpg.""" + fake_popen, _captured = self._fake_popen_capture() + fake_proc = fake_popen(["placeholder"]) + + monkeypatch.setattr("tools.process_registry._find_shell", lambda: "/bin/bash") + monkeypatch.setattr( + "tools.process_registry._systemd_run_user_scope_available", + lambda: True, + ) + monkeypatch.setattr( + "gateway.restart.is_gateway_supervisor_process", + lambda: True, + ) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run") + + broken_reader = MagicMock() + broken_reader.start.side_effect = RuntimeError("reader failed") + + with patch("subprocess.Popen", return_value=fake_proc), \ + patch("threading.Thread", return_value=broken_reader), \ + patch("tools.process_registry._stop_systemd_unit", return_value=True) as stop_unit, \ + patch("os.killpg") as killpg, \ + patch.object(registry, "_write_checkpoint"): + with pytest.raises(RuntimeError, match="reader failed"): + registry.spawn_local("echo hello", cwd="/tmp") + + stop_unit.assert_called_once() + assert stop_unit.call_args.args[0].startswith("hermes-worker-proc_") + assert stop_unit.call_args.args[0].endswith(".scope") + killpg.assert_not_called() + + def test_pty_spawn_is_wrapped_in_systemd_scope(self, registry, monkeypatch): + """Interactive executors receive the same sibling-cgroup isolation.""" + from ptyprocess import PtyProcess + + fake_pty = MagicMock() + fake_pty.pid = 4321 + + monkeypatch.setattr("tools.process_registry._find_shell", lambda: "/bin/bash") + monkeypatch.setattr( + "tools.process_registry._systemd_run_user_scope_available", + lambda: True, + ) + monkeypatch.setattr( + "gateway.restart.is_gateway_supervisor_process", + lambda: True, + ) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run") + + with patch.object(PtyProcess, "spawn", return_value=fake_pty) as pty_spawn, \ + patch("threading.Thread", return_value=MagicMock()), \ + patch.object(registry, "_write_checkpoint"): + session = registry.spawn_local("codex", cwd="/tmp", use_pty=True) + + argv = pty_spawn.call_args.args[0] + assert argv[0] == "/usr/bin/systemd-run" + assert "--scope" in argv + assert "--unit" in argv + assert "--" in argv + assert argv[-3:] == ["/bin/bash", "-lic", "set +m; codex"] + assert session.systemd_unit == f"hermes-worker-{session.id}.scope" + def test_kill_recovered_detached_already_exited_stops_persisted_scope( self, registry, monkeypatch ): @@ -1817,3 +1882,45 @@ class TestSystemdCgroupIsolation: assert not second.is_alive() assert results == [True, True] assert len(probe_calls) == 1 + + def test_failed_systemd_probe_retries_after_cache_ttl(self, monkeypatch): + import tools.process_registry as pr + + monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_AVAILABLE", None) + monkeypatch.setattr(pr, "_SYSTEMD_SCOPE_PROBED_AT", 0.0, raising=False) + clock = [100.0] + probe_results = [1, 0] + probe_calls = [] + + def fake_run(*args, **kwargs): + probe_calls.append(args) + return subprocess.CompletedProcess( + args=args[0], returncode=probe_results.pop(0) + ) + + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemd-run") + monkeypatch.setattr("tools.process_registry.time.monotonic", lambda: clock[0]) + monkeypatch.setattr("subprocess.run", fake_run) + + assert pr._systemd_run_user_scope_available() is False + assert pr._systemd_run_user_scope_available() is False + assert len(probe_calls) == 1 + + clock[0] += 61 + assert pr._systemd_run_user_scope_available() is True + assert len(probe_calls) == 2 + + def test_stop_systemd_unit_treats_absent_unit_as_clean(self, monkeypatch): + import tools.process_registry as pr + + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/systemctl") + monkeypatch.setattr( + "subprocess.run", + lambda *args, **kwargs: subprocess.CompletedProcess( + args=args[0], + returncode=5, + stderr=b"Unit hermes-worker-gone.scope not loaded.\n", + ), + ) + + assert pr._stop_systemd_unit("hermes-worker-gone.scope") is True diff --git a/tools/process_registry.py b/tools/process_registry.py index 22e3261da54b5..f29204b999124 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -97,6 +97,8 @@ WATCH_GLOBAL_COOLDOWN_SECONDS = 30 _SYSTEMD_SCOPE_AVAILABLE: Optional[bool] = None _SYSTEMD_SCOPE_PROBE_LOCK = threading.Lock() +_SYSTEMD_SCOPE_PROBED_AT = 0.0 +_SYSTEMD_SCOPE_FAILURE_TTL_SECONDS = 60.0 def _systemd_run_user_scope_available() -> bool: @@ -110,16 +112,31 @@ def _systemd_run_user_scope_available() -> bool: (``systemd-run --user --scope --unit=… -- /bin/true``) and remember the outcome. """ - global _SYSTEMD_SCOPE_AVAILABLE - if _SYSTEMD_SCOPE_AVAILABLE is not None: - return _SYSTEMD_SCOPE_AVAILABLE + global _SYSTEMD_SCOPE_AVAILABLE, _SYSTEMD_SCOPE_PROBED_AT + cached = _SYSTEMD_SCOPE_AVAILABLE + now = time.monotonic() + if cached is True: + return True + if ( + cached is False + and now - _SYSTEMD_SCOPE_PROBED_AT < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS + ): + return False # Double-checked locking keeps concurrent first-use spawns from observing # a temporary False while the definitive probe is still in flight. Such a # race would launch the losing workload back inside the gateway cgroup. with _SYSTEMD_SCOPE_PROBE_LOCK: - if _SYSTEMD_SCOPE_AVAILABLE is not None: - return _SYSTEMD_SCOPE_AVAILABLE + cached = _SYSTEMD_SCOPE_AVAILABLE + now = time.monotonic() + if cached is True: + return True + if ( + cached is False + and now - _SYSTEMD_SCOPE_PROBED_AT + < _SYSTEMD_SCOPE_FAILURE_TTL_SECONDS + ): + return False available = False if not _IS_WINDOWS: @@ -130,9 +147,7 @@ def _systemd_run_user_scope_available() -> bool: if binary: # Probe: create a transient scope that immediately exits. # A unique unit avoids collisions; timeout bounds D-Bus. - probe_unit = ( - f"hermes-probe-scope-{os.getpid()}-{int(time.time())}" - ) + probe_unit = f"hermes-probe-scope-{os.getpid()}-{uuid.uuid4().hex[:8]}" result = subprocess.run( [ binary, "--user", "--scope", "--quiet", @@ -156,6 +171,7 @@ def _systemd_run_user_scope_available() -> bool: logger.debug("systemd-run --user --scope probe error: %s", exc) _SYSTEMD_SCOPE_AVAILABLE = available + _SYSTEMD_SCOPE_PROBED_AT = time.monotonic() return available @@ -215,10 +231,17 @@ def _stop_systemd_unit(unit_name: str) -> bool: timeout=15, ) if result.returncode != 0: + stderr = (result.stderr or b"").decode(errors="replace").strip() + stderr_lower = stderr.lower() + if any( + marker in stderr_lower + for marker in ("not loaded", "not found", "does not exist") + ): + return True logger.debug( "systemctl --user stop %s exited %d: %s", unit_name, result.returncode, - result.stderr.decode(errors="replace").strip(), + stderr, ) return False return True @@ -969,7 +992,7 @@ class ProcessRegistry: # scope so it gets a separate cgroup. An OOM in the worker then # kills only the worker instead of taking down the whole gateway # cgroup (and the messaging control plane with it). We only do this - # for the common pipe-mode path; PTY mode is left as future work. + # for both pipe mode and the PTY path above. shell_argv = [user_shell, "-lic", f"set +m; {safe_command}"] use_systemd_scope = False under_supervisor = False @@ -1048,7 +1071,14 @@ class ProcessRegistry: # descendants spawned via setsid) before re-raising so they do not # leak as untracked background processes. try: - if not _IS_WINDOWS: + if session.systemd_unit: + # systemd-run --scope shares the gateway's process group + # because start_new_session=False. Never killpg here: that + # can signal the gateway itself. Stop the sibling cgroup, + # then safely terminate the wrapper PID tree as fallback. + _stop_systemd_unit(session.systemd_unit) + self._terminate_host_pid(proc.pid, session.host_start_time) + elif not _IS_WINDOWS: try: kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM) os.killpg(os.getpgid(proc.pid), kill_signal) # windows-footgun: ok - guarded by _IS_WINDOWS above