"""Checked group termination for the erlang supervisor.

Erlang runs this as its own process, so supervision keeps making progress when
the kernel cannot (a native cell holding the GIL, a wedged interpreter), and so
the supervisor never parses kill(1) error text or guesses an exit status. One
argv carries the request (or `-` and one stdin line, the form run over ssh
for a remote kernel's processes); one JSON object on stdout carries the
verdict.
"""

from __future__ import annotations

import signal
import threading

import asyncio
import json
import os
import sys

import albedo_proc
import albedo_identity

TERM_MS = 250
KILL_MS = 1000  # the supervisor's remaining budget is short: this is the last resort


def budget(request: dict, name: str, default: int) -> float:
    """Seconds for one ladder step, clamped to something a caller can wait for."""
    value = request.get(name, default)
    return (
        min(max(value, 10), default) / 1000
        if isinstance(value, int)
        else default / 1000
    )


def target(spec: object) -> albedo_proc.Group:
    """One requested target: a process group, or one process that leads none."""
    if not isinstance(spec, dict):
        raise ValueError("target must be an object")
    pgid = spec.get("pgid")
    operand = pgid if isinstance(pgid, int) else spec.get("pid")
    if not isinstance(operand, int) or operand <= 1:
        raise ValueError("target needs a pgid or pid above 1")
    leader = spec.get("leader")
    whole = isinstance(pgid, int)
    if not whole:
        try:
            whole = os.getpgid(operand) == operand
        except (ProcessLookupError, PermissionError):
            pass
    return albedo_proc.Group(
        operand, leader if isinstance(leader, str) else None, whole
    )


async def supervise_async(request: object) -> dict[str, object]:
    if not isinstance(request, dict):
        raise ValueError("request must be an object")
    if "discover" in request:
        home, known = request["discover"], request.get("known", [])
        if (
            not isinstance(home, str)
            or not isinstance(known, list)
            or not all(isinstance(path, str) for path in known)
        ):
            raise ValueError("discovery needs a home and known run directories")
        return {"kernels": albedo_identity.discover(home, known)}
    specs = request.get("targets")
    if not isinstance(specs, list):
        raise ValueError("request needs a target list")
    targets = [target(spec) for spec in specs]
    endings = await albedo_proc.terminate(
        targets,
        term=budget(request, "term_ms", TERM_MS),
        kill=budget(request, "kill_ms", KILL_MS),
    )
    return {
        "targets": [
            {**ending.as_json(), "label": spec.get("label")}
            for spec, ending in zip(specs, endings)
        ]
    }


def supervise(request: object) -> dict[str, object]:
    return asyncio.run(supervise_async(request))


async def serve() -> None:
    """Answer one JSON request per stdin line until end of input.

    The daemon keeps one of these running, so a stop does not pay for an
    interpreter start. Requests run concurrently; each answer is one stdout line
    carrying the request's id. The first line says the interpreter is ready.
    """
    loop = asyncio.get_running_loop()
    lines: asyncio.Queue[str | None] = asyncio.Queue()

    def read() -> None:
        for line in sys.stdin:
            loop.call_soon_threadsafe(lines.put_nowait, line)
        loop.call_soon_threadsafe(lines.put_nowait, None)

    threading.Thread(target=read, daemon=True).start()

    async def answer(line: str) -> None:
        identity = None
        try:
            message = json.loads(line)
            identity = message["id"]
            reply: dict[str, object] = {
                "ok": True,
                **await supervise_async(message["request"]),
            }
        except Exception as error:
            reply = {"ok": False, "error": f"{type(error).__name__}: {error}"}
        print(json.dumps({"id": identity, "answer": reply}), flush=True)

    print(json.dumps({"ready": True}), flush=True)
    running: set[asyncio.Task[None]] = set()
    while (line := await lines.get()) is not None:
        task = asyncio.create_task(answer(line))
        running.add(task)
        task.add_done_callback(running.discard)
    await asyncio.gather(*running)


def main(request: str) -> int:
    try:
        answer: dict[str, object] = {"ok": True, **supervise(json.loads(request))}
        status = 0
    except Exception as error:
        answer = {"ok": False, "error": f"{type(error).__name__}: {error}"}
        status = 2
    print(json.dumps(answer))
    return status


if __name__ == "__main__":
    if sys.argv[1:] == ["--serve"]:
        asyncio.run(serve())
        sys.exit(0)
    # Default SIGALRM terminates even a wedged interpreter; no Python handler.
    # The owning port waits longer than this before reporting a helper timeout.
    signal.signal(signal.SIGALRM, signal.SIG_DFL)
    signal.alarm(3)
    request = sys.argv[1] if len(sys.argv) > 1 else "{}"
    sys.exit(main(sys.stdin.readline() if request == "-" else request))