"""Drain one retained child session before reaping its leader; never target another process group."""
import os
import select
import signal
import subprocess
import time


class Interrupted(RuntimeError):
    pass


class Exit:
    def __init__(self, pid: int):
        self.pid, self.queue, self.done = pid, None, False
        if hasattr(os, "waitid"):
            return
        self.queue = select.kqueue()
        event = select.kevent(pid, filter=select.KQ_FILTER_PROC, flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT,
                              fflags=select.KQ_NOTE_EXIT)
        self.queue.control([event], 0, 0)

    def __call__(self) -> bool:
        if self.queue is None:
            return os.waitid(os.P_PID, self.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is not None
        if not self.done:
            events = self.queue.control(None, 1, 0)
            if events:
                if events[0].ident != self.pid or not events[0].fflags & select.KQ_NOTE_EXIT:
                    raise RuntimeError("native_pods_process_readback_failed")
                self.done = True
        return self.done

    def close(self):
        if self.queue:
            self.queue.close()


def members(pid: int) -> bool:
    result = subprocess.run(["ps", "-axo", "pid=,pgid=,stat="], capture_output=True, text=True, timeout=2, check=False)
    if result.returncode:
        raise RuntimeError("native_pods_process_readback_failed")
    for line in result.stdout.splitlines():
        fields = line.split()
        if len(fields) != 3 or not fields[0].isdigit() or not fields[1].isdigit():
            raise RuntimeError("native_pods_process_readback_failed")
        if int(fields[1]) == pid and int(fields[0]) != pid and not fields[2].startswith("Z"):
            return True
    return False


def drain(pid: int, exited: Exit) -> None:
    for sig, duration in ((signal.SIGTERM, 3), (signal.SIGKILL, 2)):
        if exited() and not members(pid):
            return
        os.killpg(pid, sig)
        deadline = time.monotonic() + duration
        while time.monotonic() < deadline:
            if exited() and not members(pid):
                return
            time.sleep(0.02)
    raise RuntimeError("native_pods_process_cleanup_unverified")


def command(argv: list[str], cwd, log, *, timeout: float = 900) -> int:
    if not hasattr(os, "waitid") and not hasattr(select, "kqueue"):
        raise RuntimeError("native_pods_process_readback_unsupported")
    child = subprocess.Popen(argv, cwd=cwd, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
    observer, interrupted = None, False
    try:
        observer = Exit(child.pid)
        deadline = time.monotonic() + timeout
        try:
            while not observer():
                if time.monotonic() >= deadline:
                    interrupted = True
                    break
                time.sleep(0.02)
        except KeyboardInterrupt:
            interrupted = True
        # WNOWAIT/kqueue retains the PID until the last group member is verified stopped.
        drain(child.pid, observer)
    except BaseException:
        # Retained PID still owns this group even when readback itself failed.
        os.killpg(child.pid, signal.SIGKILL)
        raise
    finally:
        if observer:
            observer.close()
        code = child.wait(timeout=5)
    if interrupted:
        raise Interrupted("native_pods_operation_interrupted")
    return code