"""Owned process groups: identify one, probe it, end it, report what happened.

The kernel's job supervision and the supervisor's checked signal helper both run
this ladder, so the two layers agree on what "terminated" means. A group is live
only while it holds a member we can still signal; a reaped leader or a group of
non-running (zombie) members is not work.
"""

from __future__ import annotations

from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import Any
import asyncio
import ctypes
import os
import signal
import struct
import subprocess
import sys

TERM_GRACE = 0.25  # seconds a group gets to honour TERM
KILL_GRACE = 2.0  # seconds a group gets to die after KILL
PROBE_INTERVAL = 0.02
LIBC = ctypes.CDLL(None)


def reap_stopped_safely(loop: asyncio.AbstractEventLoop) -> None:
    """Let `loop` run children that get paused with SIGSTOP.

    Python 3.14's threaded child watcher waits with waitid(WEXITED | WNOWAIT),
    which macOS also answers for a stopped child, and then reaps with a blocking
    waitpid on the event loop: a job paused for a heavy slot froze the kernel.
    The replacement reaps in its thread with a waitpid that ignores stops, as
    3.13 did. Linux uses pidfds, which only wake on exit, and needs nothing.
    """
    threaded: Any = getattr(
        getattr(asyncio, "unix_events", None), "_ThreadedChildWatcher", None
    )
    if (
        sys.platform != "darwin"
        or threaded is None
        or not isinstance(getattr(loop, "_watcher", None), threaded)
    ):
        return

    class ExitWatcher(threaded):
        def _do_waitpid(
            self,
            loop: asyncio.AbstractEventLoop,
            expected_pid: int,
            callback: Callable[..., object],
            args: tuple[object, ...],
        ) -> None:
            try:
                returncode = os.waitstatus_to_exitcode(os.waitpid(expected_pid, 0)[1])
            except ChildProcessError:
                returncode = 255
            try:
                loop.call_soon_threadsafe(callback, expected_pid, returncode, *args)
            except RuntimeError:  # the loop closed first
                pass
            self._threads.pop(expected_pid, None)

    setattr(loop, "_watcher", ExitWatcher())


@dataclass(frozen=True)
class Group:
    """A target to end: a whole process group, or one process that leads none.

    `leader` identifies the leader by its start time, so a reused id is refused.
    """

    pgid: int
    leader: str | None = None
    whole: bool = True


@dataclass(frozen=True)
class Termination:
    """What one termination attempt established about one process group."""

    pgid: int | None
    signals: tuple[str, ...]
    gone: bool
    failures: tuple[str, ...] = ()
    note: str = ""

    def report(self) -> str:
        where = (
            "no process group" if self.pgid is None else f"process group {self.pgid}"
        )
        outcome = "terminated" if self.gone else "SURVIVED"
        sent = "+".join(self.signals) if self.signals else "no signal"
        detail = (
            "; ".join((*self.failures, self.note))
            if self.note
            else "; ".join(self.failures)
        )
        return f"{where} {outcome} after {sent}" + (f": {detail}" if detail else "")

    def as_json(self) -> dict[str, object]:
        return {
            "pgid": self.pgid,
            "signals": list(self.signals),
            "gone": self.gone,
            "failures": list(self.failures),
            "note": self.note,
        }


def start_token(stat: bytes) -> str | None:
    """Start time of one /proc/<pid>/stat line (field 22, after the command name)."""
    try:
        return stat.rsplit(b")", 1)[1].split()[19].decode()
    except (IndexError, UnicodeDecodeError):
        return None


def leader_token(pid: int) -> str | None:
    """Identity token for a live pid: its start time, from /proc or, on
    Darwin, its process table entry. None where neither can tell."""
    try:
        with open(f"/proc/{pid}/stat", "rb") as stat:
            return start_token(stat.read())
    except OSError:
        return darwin_start(pid) if sys.platform == "darwin" else None


def darwin_start(pid: int) -> str | None:
    """A Darwin process's start time: the timeval that opens its kinfo_proc
    (sysctl CTL_KERN, KERN_PROC, KERN_PROC_PID). None for a pid not in use."""
    mib = (ctypes.c_int * 4)(1, 14, 1, pid)
    buffer = ctypes.create_string_buffer(1024)
    size = ctypes.c_size_t(len(buffer))
    if LIBC.sysctl(mib, 4, buffer, ctypes.byref(size), None, 0) != 0 or size.value < 12:
        return None
    seconds, micros = struct.unpack_from("=qi", buffer.raw)
    return f"{seconds}.{micros:06d}"


def live_members(pgid: int) -> list[int] | None:
    """Pids in a group that are not zombies; None where /proc is unavailable."""
    try:
        entries = os.listdir("/proc")
    except OSError:
        if sys.platform != "darwin":
            return None
        # Darwin has no /proc. Ask its process table; EPERM alone never means dead.
        try:
            result = subprocess.run(
                ["/bin/ps", "-A", "-o", "pid=", "-o", "pgid=", "-o", "stat="],
                capture_output=True,
                text=True,
                timeout=0.5,
                check=True,
            )
            rows = [line.split() for line in result.stdout.splitlines()]
            return [
                int(pid)
                for pid, group, state in rows
                if int(group) == pgid and not state.startswith("Z")
            ]
        except (OSError, subprocess.SubprocessError, ValueError):
            return None
    members = []
    for entry in entries:
        if not entry.isdigit():
            continue
        try:
            with open(f"/proc/{entry}/stat", "rb") as stat:
                fields = stat.read().rsplit(b")", 1)[1].split()
        except FileNotFoundError:
            continue  # process exited while enumerating
        except (OSError, IndexError):
            return None  # an incomplete view cannot prove the group empty
        if fields[0] == b"Z" or fields[2] != str(pgid).encode():
            continue
        members.append(int(entry))
    return members


def deliver(group: Group, sig: int) -> None:
    """Signal the target: its group, or the single process when it leads none."""
    (os.killpg if group.whole else os.kill)(group.pgid, sig)


def alive(group: Group) -> bool:
    """True while the target exists and still looks like the one we created."""
    try:
        deliver(group, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # existence is known; inability to signal is a cleanup failure
    if group.leader is None:
        return True
    token = leader_token(group.pgid)
    return token is None or token == group.leader


def current(group: Group) -> bool:
    """alive/1 refined by membership: zombies awaiting reaping are not work."""
    if not alive(group):
        return False
    if not group.whole:
        return True
    members = live_members(group.pgid)
    return True if members is None else bool(members)


def send_signal(group: Group, sig: int) -> str | None:
    """Send one signal to a target; returns the failure text when it did not land."""
    try:
        deliver(group, sig)
        return None
    except ProcessLookupError:
        return None  # already gone; the probe decides
    except OSError as error:
        return f"{signal.Signals(sig).name} to {group.pgid}: {error.strerror or error}"


async def settled(live: dict[int, Group], window: float) -> dict[int, Group]:
    """The groups still alive after one shared wait, so batches share a deadline."""
    deadline = asyncio.get_running_loop().time() + window
    while True:
        remaining = {pgid: group for pgid, group in live.items() if alive(group)}
        if not remaining or asyncio.get_running_loop().time() >= deadline:
            return remaining
        await asyncio.sleep(PROBE_INTERVAL)


async def terminate(
    groups: Sequence[Group], term: float = TERM_GRACE, kill: float = KILL_GRACE
) -> list[Termination]:
    """End every group: TERM, then KILL, one shared deadline per step."""
    live = {group.pgid: group for group in groups if alive(group)}
    sent: dict[int, list[str]] = {pgid: [] for pgid in live}
    failures: dict[int, list[str]] = {pgid: [] for pgid in live}
    for sig, window in ((signal.SIGTERM, term), (signal.SIGKILL, kill)):
        for pgid, group in live.items():
            failure = send_signal(group, sig)
            sent[pgid].append(signal.Signals(sig).name)
            if failure is not None:
                failures[pgid].append(failure)
        live = await settled(live, window)
        if not live:
            break
    endings = []
    for group in groups:
        pgid, signals = group.pgid, sent.get(group.pgid, [])
        surviving, note = pgid in live, ""
        if surviving and not current(group):
            surviving, note = False, "only non-running members remained"
        endings.append(
            Termination(
                pgid, tuple(signals), not surviving, tuple(failures.get(pgid, [])), note
            )
        )
    return endings