Jack1808/Claude_Code

1"""Track and clean up spawned CLI subprocesses.2 3This is a safety net for cases where the server is interrupted (Ctrl+C) and the4FastAPI lifespan cleanup doesn't run to completion. We only track processes we5spawn so we don't accidentally kill unrelated system processes.6"""7 8from __future__ import annotations9 10import atexit11import os12import subprocess13import threading14 15from loguru import logger16 17_lock = threading.Lock()18_pids: set[int] = set()19_atexit_registered = False20 21 22def ensure_atexit_registered() -> None:23    global _atexit_registered24    with _lock:25        if _atexit_registered:26            return27        atexit.register(kill_all_best_effort)28        _atexit_registered = True29 30 31def register_pid(pid: int) -> None:32    if not pid:33        return34    ensure_atexit_registered()35    with _lock:36        _pids.add(int(pid))37 38 39def unregister_pid(pid: int) -> None:40    if not pid:41        return42    with _lock:43        _pids.discard(int(pid))44 45 46def kill_all_best_effort() -> None:47    """Kill any still-running registered pids (best-effort)."""48    with _lock:49        pids = list(_pids)50        _pids.clear()51 52    if not pids:53        return54 55    if os.name == "nt":56        for pid in pids:57            try:58                # /T kills child processes, /F forces termination.59                subprocess.run(60                    ["taskkill", "/PID", str(pid), "/T", "/F"],61                    stdout=subprocess.DEVNULL,62                    stderr=subprocess.DEVNULL,63                    check=False,64                )65            except Exception as e:66                logger.debug("process_registry: taskkill failed pid=%s: %s", pid, e)67        return68 69    # Best-effort fallback for non-Windows.70    for pid in pids:71        try:72            os.kill(pid, 9)73        except Exception as e:74            logger.debug("process_registry: kill failed pid=%s: %s", pid, e)75