From a28d39f80f27d5227f8a2464957aa1ce3c6250e8 Mon Sep 17 00:00:00 2001 From: Nigmat Rahim <174248627+Nigmat-future@users.noreply.github.com> Date: Fri, 19 Jun 2026 05:42:00 +0800 Subject: [PATCH 1/3] fix(tui): reap orphaned slash_worker processes on gateway startup Orphaned tui_gateway.slash_worker subprocesses survived gateway restarts and blocked Desktop reconnects. Reap unowned workers at startup, bind new workers to a Windows kill-on-close job object, and treat zombie parents as orphaned in the in-worker watchdog. Fixes #48643 --- tests/test_slash_worker_watchdog.py | 16 ++ .../test_slash_worker_lifecycle.py | 163 ++++++++++++++ tui_gateway/server.py | 7 + tui_gateway/slash_worker.py | 5 +- tui_gateway/slash_worker_lifecycle.py | 213 ++++++++++++++++++ 5 files changed, 403 insertions(+), 1 deletion(-) create mode 100644 tests/tui_gateway/test_slash_worker_lifecycle.py create mode 100644 tui_gateway/slash_worker_lifecycle.py diff --git a/tests/test_slash_worker_watchdog.py b/tests/test_slash_worker_watchdog.py index 198524c522dc..9f88848eb81c 100644 --- a/tests/test_slash_worker_watchdog.py +++ b/tests/test_slash_worker_watchdog.py @@ -1,4 +1,5 @@ import psutil +import pytest from tui_gateway import slash_worker @@ -19,3 +20,18 @@ def test_is_orphaned_false_when_parent_alive_and_matches(): assert ( slash_worker._is_orphaned(me.pid, me.create_time(), getppid=lambda: me.pid) is False ) + + +def test_is_orphaned_true_when_parent_is_zombie(monkeypatch): + me = psutil.Process() + + class ZombieParent: + def status(self): + return psutil.STATUS_ZOMBIE + + def create_time(self): + return me.create_time() + + monkeypatch.setattr(slash_worker.psutil, "pid_exists", lambda pid: True) + monkeypatch.setattr(slash_worker.psutil, "Process", lambda pid: ZombieParent()) + assert slash_worker._is_orphaned(me.pid, me.create_time(), getppid=lambda: me.pid) is True diff --git a/tests/tui_gateway/test_slash_worker_lifecycle.py b/tests/tui_gateway/test_slash_worker_lifecycle.py new file mode 100644 index 000000000000..66cf8d4da2ae --- /dev/null +++ b/tests/tui_gateway/test_slash_worker_lifecycle.py @@ -0,0 +1,163 @@ +"""Tests for slash_worker orphan reaping and lifecycle helpers.""" + +from __future__ import annotations + +import psutil +import pytest + +from tui_gateway import slash_worker_lifecycle as lifecycle + + +def test_is_slash_worker_cmdline(): + assert lifecycle.is_slash_worker_cmdline( + ["/usr/bin/python", "-m", "tui_gateway.slash_worker", "--session-key", "k"] + ) + assert not lifecycle.is_slash_worker_cmdline(["python", "-m", "pytest"]) + + +def test_has_live_gateway_owner_true_for_direct_child(monkeypatch): + my_pid = 4242 + + class FakeParent: + ppid = my_pid + + def cmdline(self): + return [] + + monkeypatch.setattr(lifecycle.psutil, "Process", lambda pid: FakeParent()) + assert lifecycle.has_live_gateway_owner(9001, my_pid=my_pid) is True + + +def test_has_live_gateway_owner_true_for_gateway_ancestor(monkeypatch): + gateway_pid = 5000 + worker_pid = 9001 + + class GatewayProc: + ppid = 1 + + def cmdline(self): + return ["python", "-m", "tui_gateway.entry"] + + def status(self): + return psutil.STATUS_RUNNING + + class WorkerParent: + ppid = gateway_pid + + def cmdline(self): + return ["conhost.exe"] + + def status(self): + return psutil.STATUS_RUNNING + + def fake_process(pid): + if pid == worker_pid: + return WorkerParent() + if pid == gateway_pid: + return GatewayProc() + raise psutil.NoSuchProcess(pid) + + monkeypatch.setattr(lifecycle.psutil, "Process", fake_process) + assert lifecycle.has_live_gateway_owner(worker_pid, my_pid=4242) is True + + +def test_has_live_gateway_owner_false_when_chain_breaks(monkeypatch): + worker_pid = 9001 + + class WorkerParent: + ppid = 7777 + + def cmdline(self): + return ["bash"] + + def status(self): + return psutil.STATUS_RUNNING + + def fake_process(pid): + if pid == worker_pid: + return WorkerParent() + raise psutil.NoSuchProcess(pid) + + monkeypatch.setattr(lifecycle.psutil, "Process", fake_process) + assert lifecycle.has_live_gateway_owner(worker_pid, my_pid=4242) is False + + +def test_reap_orphan_slash_workers_terminates_unowned(monkeypatch): + terminated: list[int] = [] + + class FakeProcInfo: + def __init__(self, pid, cmdline): + self.info = {"pid": pid, "cmdline": cmdline} + + rows = [ + FakeProcInfo(100, ["python", "-m", "tui_gateway.slash_worker", "--session-key", "k1"]), + FakeProcInfo(101, ["python", "-m", "pytest"]), + ] + + monkeypatch.setattr( + lifecycle.psutil, + "process_iter", + lambda attrs: iter(rows), + ) + monkeypatch.setattr( + lifecycle, + "has_live_gateway_owner", + lambda pid, my_pid: False, + ) + monkeypatch.setattr( + lifecycle, + "_terminate_pid", + lambda pid: terminated.append(pid), + ) + + count = lifecycle.reap_orphan_slash_workers(my_pid=4242) + assert count == 1 + assert terminated == [100] + + +def test_reap_orphan_slash_workers_skips_owned_workers(monkeypatch): + terminated: list[int] = [] + + class FakeProcInfo: + def __init__(self, pid, cmdline): + self.info = {"pid": pid, "cmdline": cmdline} + + rows = [ + FakeProcInfo(100, ["python", "-m", "tui_gateway.slash_worker", "--session-key", "k1"]), + ] + + monkeypatch.setattr( + lifecycle.psutil, + "process_iter", + lambda attrs: iter(rows), + ) + monkeypatch.setattr( + lifecycle, + "has_live_gateway_owner", + lambda pid, my_pid: True, + ) + monkeypatch.setattr( + lifecycle, + "_terminate_pid", + lambda pid: terminated.append(pid), + ) + + count = lifecycle.reap_orphan_slash_workers(my_pid=4242) + assert count == 0 + assert terminated == [] + + +def test_maybe_reap_orphan_slash_workers_on_startup_runs_once(monkeypatch): + lifecycle._reaper_ran = False + calls: list[int] = [] + + monkeypatch.setattr( + lifecycle, + "reap_orphan_slash_workers", + lambda **kwargs: calls.append(1) or 2, + ) + + lifecycle.maybe_reap_orphan_slash_workers_on_startup() + lifecycle.maybe_reap_orphan_slash_workers_on_startup() + + assert calls == [1] diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 294e543c230f..b9d08d664b1e 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -257,6 +257,9 @@ def __init__(self, session_key: str, model: str): cwd=os.getcwd(), env=os.environ.copy(), ) + from tui_gateway.slash_worker_lifecycle import attach_slash_worker_kill_job + + self._win_kill_job = attach_slash_worker_kill_job(self.proc) threading.Thread(target=self._drain_stdout, daemon=True).start() threading.Thread(target=self._drain_stderr, daemon=True).start() @@ -640,6 +643,10 @@ def _loop(): atexit.register(_shutdown_sessions) _start_idle_reaper() +from tui_gateway.slash_worker_lifecycle import maybe_reap_orphan_slash_workers_on_startup + +maybe_reap_orphan_slash_workers_on_startup() + # ── Plumbing ────────────────────────────────────────────────────────── diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py index fce8ec3e26b2..740dd2b40797 100644 --- a/tui_gateway/slash_worker.py +++ b/tui_gateway/slash_worker.py @@ -47,7 +47,10 @@ def _is_orphaned(original_ppid, parent_create_time, getppid=os.getppid) -> bool: try: if not psutil.pid_exists(original_ppid): return True - return psutil.Process(original_ppid).create_time() != parent_create_time + parent = psutil.Process(original_ppid) + if parent.status() in (psutil.STATUS_ZOMBIE, psutil.STATUS_DEAD): + return True + return parent.create_time() != parent_create_time except psutil.Error: return True diff --git a/tui_gateway/slash_worker_lifecycle.py b/tui_gateway/slash_worker_lifecycle.py new file mode 100644 index 000000000000..dbaae0c5c5aa --- /dev/null +++ b/tui_gateway/slash_worker_lifecycle.py @@ -0,0 +1,213 @@ +"""Lifecycle helpers for ``tui_gateway.slash_worker`` subprocesses. + +Orphaned slash workers (gateway crash, hard kill mid-compaction, session +reaped without worker.close) block Desktop reconnects until manually killed. +This module reaps them on gateway startup and ties new workers to the gateway +process lifetime on Windows via a kill-on-close job object. +""" + +from __future__ import annotations + +import logging +import os +import subprocess +import sys +from typing import Iterable + +import psutil + +logger = logging.getLogger(__name__) + +_SLASH_WORKER_MARKER = "tui_gateway.slash_worker" +_GATEWAY_CMD_MARKERS = ( + "tui_gateway.entry", + "tui_gateway.server", + "tui_gateway", + "hermes_cli.main gateway", + "hermes gateway", +) + +_reaper_ran = False + + +def _cmdline_text(cmdline: Iterable[str] | None) -> str: + if not cmdline: + return "" + return " ".join(str(part) for part in cmdline).lower() + + +def is_slash_worker_cmdline(cmdline: Iterable[str] | None) -> bool: + text = _cmdline_text(cmdline) + return _SLASH_WORKER_MARKER in text + + +def is_gateway_cmdline(cmdline: Iterable[str] | None) -> bool: + text = _cmdline_text(cmdline) + return any(marker in text for marker in _GATEWAY_CMD_MARKERS) + + +def has_live_gateway_owner(worker_pid: int, *, my_pid: int) -> bool: + """True when ``worker_pid`` descends from a live gateway (this or another).""" + try: + current = psutil.Process(worker_pid).ppid + except (psutil.NoSuchProcess, psutil.AccessDenied): + return False + + seen: set[int] = set() + while current and current not in seen: + seen.add(current) + if current == my_pid: + return True + try: + parent = psutil.Process(current) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return False + status = parent.status() + if status in (psutil.STATUS_ZOMBIE, psutil.STATUS_DEAD): + return False + if is_gateway_cmdline(parent.cmdline()): + return True + current = parent.ppid + return False + + +def _terminate_pid(pid: int) -> None: + try: + proc = psutil.Process(pid) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return + try: + proc.terminate() + proc.wait(timeout=1) + return + except (psutil.TimeoutExpired, psutil.NoSuchProcess, psutil.AccessDenied): + pass + try: + proc.kill() + proc.wait(timeout=1) + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.TimeoutExpired): + pass + + +def reap_orphan_slash_workers(*, my_pid: int | None = None) -> int: + """Terminate slash_worker processes with no live gateway owner. + + Safe to call at gateway startup and idempotent across concurrent gateway + instances: workers owned by another live ``tui_gateway`` process are kept. + """ + owner_pid = my_pid or os.getpid() + reaped = 0 + try: + processes = psutil.process_iter(["pid", "cmdline"]) + except Exception as exc: + logger.debug("slash_worker orphan reaper: process_iter failed: %s", exc) + return 0 + + for info in processes: + try: + pid = int(info.info["pid"]) + except (TypeError, ValueError, KeyError): + continue + if pid == owner_pid: + continue + cmdline = info.info.get("cmdline") or [] + if not is_slash_worker_cmdline(cmdline): + continue + if has_live_gateway_owner(pid, my_pid=owner_pid): + continue + _terminate_pid(pid) + reaped += 1 + logger.info( + "Reaped orphaned slash_worker PID %d (cmd=%r)", + pid, + " ".join(cmdline) if cmdline else "", + ) + return reaped + + +def maybe_reap_orphan_slash_workers_on_startup() -> None: + """Run the orphan reaper once per gateway process.""" + global _reaper_ran + if _reaper_ran: + return + _reaper_ran = True + try: + count = reap_orphan_slash_workers() + except Exception: + logger.debug("slash_worker orphan reaper failed", exc_info=True) + return + if count: + logger.info("Reaped %d orphaned slash_worker process(es) on startup", count) + + +def attach_slash_worker_kill_job(proc: subprocess.Popen) -> object | None: + """On Windows, bind the worker to a job that dies with this process.""" + if sys.platform != "win32": + return None + handle = getattr(proc, "_handle", None) + if handle is None: + return None + try: + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000 + JobObjectExtendedLimitInformation = 9 + + class IO_COUNTERS(ctypes.Structure): + _fields_ = [ + ("ReadOperationCount", ctypes.c_uint64), + ("WriteOperationCount", ctypes.c_uint64), + ("OtherOperationCount", ctypes.c_uint64), + ("ReadTransferCount", ctypes.c_uint64), + ("WriteTransferCount", ctypes.c_uint64), + ("OtherTransferCount", ctypes.c_uint64), + ] + + class JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure): + _fields_ = [ + ("PerProcessUserTimeLimit", ctypes.c_int64), + ("PerJobUserTimeLimit", ctypes.c_int64), + ("LimitFlags", wintypes.DWORD), + ("MinimumWorkingSetSize", ctypes.c_size_t), + ("MaximumWorkingSetSize", ctypes.c_size_t), + ("ActiveProcessLimit", wintypes.DWORD), + ("Affinity", ctypes.c_size_t), + ("PriorityClass", wintypes.DWORD), + ("SchedulingClass", wintypes.DWORD), + ] + + class JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure): + _fields_ = [ + ("BasicLimitInformation", JOBOBJECT_BASIC_LIMIT_INFORMATION), + ("IoInfo", IO_COUNTERS), + ("ProcessMemoryLimit", ctypes.c_size_t), + ("JobMemoryLimit", ctypes.c_size_t), + ("PeakProcessMemoryUsed", ctypes.c_size_t), + ("PeakJobMemoryUsed", ctypes.c_size_t), + ] + + job = kernel32.CreateJobObjectW(None, None) + if not job: + return None + + info = JOBOBJECT_EXTENDED_LIMIT_INFORMATION() + info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if not kernel32.SetInformationJobObject( + job, + JobObjectExtendedLimitInformation, + ctypes.byref(info), + ctypes.sizeof(info), + ): + kernel32.CloseHandle(job) + return None + + if not kernel32.AssignProcessToJobObject(job, wintypes.HANDLE(int(handle))): + kernel32.CloseHandle(job) + return None + return job + except Exception: + logger.debug("Failed to attach slash_worker to kill-on-close job", exc_info=True) + return None From 75dbba7902587b527eacd6399de1aa06288f7dfc Mon Sep 17 00:00:00 2001 From: Nigmat Rahim <174248627+Nigmat-future@users.noreply.github.com> Date: Fri, 19 Jun 2026 05:49:21 +0800 Subject: [PATCH 2/3] fix(tui): call Process.ppid() for psutil 7.x compatibility psutil 7.x exposes ppid as a method; attribute access returned a bound method and crashed the orphan reaper on real process scans. --- .../test_slash_worker_lifecycle.py | 36 +++++++++++++++++++ tui_gateway/slash_worker_lifecycle.py | 12 +++++-- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/tests/tui_gateway/test_slash_worker_lifecycle.py b/tests/tui_gateway/test_slash_worker_lifecycle.py index 66cf8d4da2ae..15c42b4614dc 100644 --- a/tests/tui_gateway/test_slash_worker_lifecycle.py +++ b/tests/tui_gateway/test_slash_worker_lifecycle.py @@ -82,6 +82,42 @@ def fake_process(pid): assert lifecycle.has_live_gateway_owner(worker_pid, my_pid=4242) is False +def test_has_live_gateway_owner_supports_psutil7_ppid_method(monkeypatch): + my_pid = 4242 + worker_pid = 9001 + gateway_pid = 5000 + + class GatewayProc: + def ppid(self): + return 1 + + def cmdline(self): + return ["python", "-m", "tui_gateway.entry"] + + def status(self): + return psutil.STATUS_RUNNING + + class WorkerParent: + def ppid(self): + return gateway_pid + + def cmdline(self): + return ["conhost.exe"] + + def status(self): + return psutil.STATUS_RUNNING + + def fake_process(pid): + if pid == worker_pid: + return WorkerParent() + if pid == gateway_pid: + return GatewayProc() + raise psutil.NoSuchProcess(pid) + + monkeypatch.setattr(lifecycle.psutil, "Process", fake_process) + assert lifecycle.has_live_gateway_owner(worker_pid, my_pid=my_pid) is True + + def test_reap_orphan_slash_workers_terminates_unowned(monkeypatch): terminated: list[int] = [] diff --git a/tui_gateway/slash_worker_lifecycle.py b/tui_gateway/slash_worker_lifecycle.py index dbaae0c5c5aa..0be8616e8bfe 100644 --- a/tui_gateway/slash_worker_lifecycle.py +++ b/tui_gateway/slash_worker_lifecycle.py @@ -30,6 +30,14 @@ _reaper_ran = False +def _process_ppid(proc: psutil.Process) -> int: + """Return parent PID across psutil versions (property on 5.x, method on 7.x).""" + ppid = proc.ppid + if callable(ppid): + return int(ppid()) + return int(ppid) + + def _cmdline_text(cmdline: Iterable[str] | None) -> str: if not cmdline: return "" @@ -49,7 +57,7 @@ def is_gateway_cmdline(cmdline: Iterable[str] | None) -> bool: def has_live_gateway_owner(worker_pid: int, *, my_pid: int) -> bool: """True when ``worker_pid`` descends from a live gateway (this or another).""" try: - current = psutil.Process(worker_pid).ppid + current = _process_ppid(psutil.Process(worker_pid)) except (psutil.NoSuchProcess, psutil.AccessDenied): return False @@ -67,7 +75,7 @@ def has_live_gateway_owner(worker_pid: int, *, my_pid: int) -> bool: return False if is_gateway_cmdline(parent.cmdline()): return True - current = parent.ppid + current = _process_ppid(parent) return False From 7c980a71fcdcb39de3c0e55c821ef676448e6783 Mon Sep 17 00:00:00 2001 From: Nigmat Rahim <174248627+Nigmat-future@users.noreply.github.com> Date: Fri, 19 Jun 2026 05:56:23 +0800 Subject: [PATCH 3/3] fix(tui): break slash_worker out of parent job for kill-on-close binding Use CREATE_BREAKAWAY_FROM_JOB when spawning slash workers on Windows so AssignProcessToJobObject succeeds under Desktop/Electron parent jobs. Add Windows e2e tests for job teardown and orphan reaping. --- .../test_slash_worker_lifecycle_e2e.py | 79 +++++++++++++++++++ tui_gateway/server.py | 7 ++ 2 files changed, 86 insertions(+) create mode 100644 tests/tui_gateway/test_slash_worker_lifecycle_e2e.py diff --git a/tests/tui_gateway/test_slash_worker_lifecycle_e2e.py b/tests/tui_gateway/test_slash_worker_lifecycle_e2e.py new file mode 100644 index 000000000000..6ad12bf85dcf --- /dev/null +++ b/tests/tui_gateway/test_slash_worker_lifecycle_e2e.py @@ -0,0 +1,79 @@ +"""Windows-only end-to-end checks for slash_worker lifecycle helpers.""" + +from __future__ import annotations + +import subprocess +import sys +import time + +import psutil +import pytest + +pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="Windows job-object e2e") + + +def test_kill_on_close_job_terminates_child_when_spawner_exits(): + script = """ +import subprocess, sys +from tui_gateway.slash_worker_lifecycle import attach_slash_worker_kill_job + +BREAKAWAY = 0x01000000 +proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(120)"], + creationflags=BREAKAWAY, +) +job = attach_slash_worker_kill_job(proc) +if job is None: + raise SystemExit("job attach failed") +# Keep job handle alive until this process exits. +_KEEP_JOB = job +print(proc.pid, flush=True) +""" + spawner = subprocess.Popen( + [sys.executable, "-c", script], + stdout=subprocess.PIPE, + text=True, + ) + out, _ = spawner.communicate(timeout=15) + worker_pid = int(out.strip()) + assert spawner.returncode == 0 + time.sleep(2) + assert not psutil.pid_exists(worker_pid), "child should die when job handle closes" + + +def test_reaper_kills_orphan_slash_worker_after_launcher_exits(): + launcher = """ +import subprocess, sys, time +BREAKAWAY = 0x01000000 +proc = subprocess.Popen( + [sys.executable, "-m", "tui_gateway.slash_worker", "--session-key", "e2e-reap"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + creationflags=BREAKAWAY, +) +print(proc.pid, flush=True) +time.sleep(0.5) +""" + launcher_proc = subprocess.Popen( + [sys.executable, "-c", launcher], + stdout=subprocess.PIPE, + text=True, + ) + out, _ = launcher_proc.communicate(timeout=20) + worker_pid = int(out.strip()) + # Give watchdog a moment; if still alive, reaper must kill it. + time.sleep(1) + if not psutil.pid_exists(worker_pid): + pytest.skip("watchdog already reaped worker before reaper ran") + + from tui_gateway.slash_worker_lifecycle import reap_orphan_slash_workers + + reaped = reap_orphan_slash_workers(my_pid=__import__("os").getpid()) + time.sleep(1) + try: + assert reaped >= 1 + assert not psutil.pid_exists(worker_pid) + finally: + if psutil.pid_exists(worker_pid): + psutil.Process(worker_pid).kill() diff --git a/tui_gateway/server.py b/tui_gateway/server.py index b9d08d664b1e..3325bc6efa80 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -246,6 +246,12 @@ def __init__(self, session_key: str, model: str): if model: argv += ["--model", model] + popen_kwargs: dict = {} + if sys.platform == "win32": + # Escape the parent job (Electron/Desktop) so AssignProcessToJobObject + # can bind the worker to our kill-on-close job (#48643). + popen_kwargs["creationflags"] = 0x01000000 # CREATE_BREAKAWAY_FROM_JOB + self._closed = False self.proc = subprocess.Popen( argv, @@ -256,6 +262,7 @@ def __init__(self, session_key: str, model: str): bufsize=1, cwd=os.getcwd(), env=os.environ.copy(), + **popen_kwargs, ) from tui_gateway.slash_worker_lifecycle import attach_slash_worker_kill_job