"""Process control for robustness cases: isolated env, timeouts, kill -9.

Only PIDs this module starts are ever signalled (never pkill by pattern).

"""

from __future__ import annotations

import os

import platform

import shutil

import signal

import socket

import subprocess

import threading

import time

from dataclasses import dataclass, field

from pathlib import Path

from typing import Callable, Optional

from . import REPO_ROOT  # noqa: F401  (ensures scripts/ on sys.path)

from oakbench.environment import base_env

DEFAULT_TIMEOUT_S = 90.0

@dataclass

class Result:

    argv: list[str]

    returncode: Optional[int]  # None = timed out (killed by harness)

    stdout: str

    stderr: str

    elapsed_ms: float

    killed_by_harness: bool = False

    timed_out: bool = False

    @property

    def ok(self) -> bool:

        return self.returncode == 0

    def text(self) -> str:

        return (self.stdout or "") + "\n" + (self.stderr or "")

    def brief(self, limit: int = 600) -> dict:

        return {

            "argv": self.argv,

            "returncode": self.returncode,

            "elapsed_ms": round(self.elapsed_ms, 3),

            "killed_by_harness": self.killed_by_harness,

            "timed_out": self.timed_out,

            "stdout_tail": (self.stdout or "")[-limit:],

            "stderr_tail": (self.stderr or "")[-limit:],

        }

def subject_env(home: Path, extra: Optional[dict] = None) -> dict:

    """Isolated HOME + deterministic identity; no inherited Oak remote/repo."""

    home.mkdir(parents=True, exist_ok=True)

    env = base_env(home=home)

    for key in ("OAK_REMOTE", "OAK_REPO", "OAK_TOKEN", "OAK_API_KEY", "DYLD_INSERT_LIBRARIES",

                "OAK_LOG", "OAK_VERBOSE", "OAK_PROGRESS"):

        env.pop(key, None)

    env["OAK_PROGRESS"] = "never"

    if extra:

        env.update(extra)

    return env

def run(argv: list[str], cwd: Path, env: dict, timeout: float = DEFAULT_TIMEOUT_S,

        stdin: Optional[str] = None) -> Result:

    start = time.perf_counter()

    proc = subprocess.Popen(argv, cwd=str(cwd), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                            stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,

                            start_new_session=True)

    try:

        out, err = proc.communicate(input=stdin.encode() if stdin is not None else None, timeout=timeout)

        timed_out = False

    except subprocess.TimeoutExpired:

        _kill_group(proc.pid)

        out, err = proc.communicate()

        timed_out = True

    elapsed = (time.perf_counter() - start) * 1000.0

    return Result(argv, None if timed_out else proc.returncode, out.decode("utf-8", "replace"),

                  err.decode("utf-8", "replace"), elapsed, timed_out=timed_out, killed_by_harness=timed_out)

def _kill_group(pid: int) -> None:

    try:

        os.killpg(pid, signal.SIGKILL)

    except (ProcessLookupError, PermissionError):

        try:

            os.kill(pid, signal.SIGKILL)

        except ProcessLookupError:

            pass

def run_killed_after(argv: list[str], cwd: Path, env: dict, delay_s: float,

                     timeout: float = DEFAULT_TIMEOUT_S, whole_group: bool = False) -> Result:

    """Start argv and SIGKILL it after delay_s (time-based crash point).

    whole_group=False kills only the client PID: children (e.g. Git's local

    receive-pack) survive, like a server that keeps going after its client died.

    """

    start = time.perf_counter()

    proc = subprocess.Popen(argv, cwd=str(cwd), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                            stdin=subprocess.DEVNULL, start_new_session=True)

    killed = threading.Event()

    def killer() -> None:

        if proc.poll() is None:

            try:

                if whole_group:

                    os.killpg(proc.pid, signal.SIGKILL)

                else:

                    os.kill(proc.pid, signal.SIGKILL)

                killed.set()

            except ProcessLookupError:

                pass

    timer = threading.Timer(delay_s, killer)

    timer.start()

    try:

        out, err = proc.communicate(timeout=timeout)

        timed_out = False

    except subprocess.TimeoutExpired:

        _kill_group(proc.pid)

        out, err = proc.communicate()

        timed_out = True

    finally:

        timer.cancel()

    elapsed = (time.perf_counter() - start) * 1000.0

    rc = None if timed_out else proc.returncode

    return Result(argv, rc, out.decode("utf-8", "replace"), err.decode("utf-8", "replace"), elapsed,

                  killed_by_harness=killed.is_set() or timed_out, timed_out=timed_out)

def free_port() -> int:

    with socket.socket() as sock:

        sock.bind(("127.0.0.1", 0))

        return int(sock.getsockname()[1])

def copy_tree(src: Path, dst: Path) -> None:

    if dst.exists():

        shutil.rmtree(dst)

    if platform.system() == "Darwin":

        # APFS clonefile copy: byte-identical, copy-on-write, ~10x faster per trial.

        proc = subprocess.run(["/bin/cp", "-cR", str(src), str(dst)], capture_output=True)

        if proc.returncode == 0:

            return

        shutil.rmtree(dst, ignore_errors=True)

    shutil.copytree(src, dst, symlinks=True)

def wait_until(predicate: Callable[[], bool], timeout: float, interval: float = 0.05) -> bool:

    deadline = time.monotonic() + timeout

    while time.monotonic() < deadline:

        if predicate():

            return True

        time.sleep(interval)

    return predicate()

@dataclass

class Spawned:

    """A long-lived helper process we own (loopback serve)."""

    proc: subprocess.Popen

    log_path: Path

    extra: dict = field(default_factory=dict)

    def stop(self) -> None:

        if self.proc.poll() is None:

            self.proc.terminate()

            try:

                self.proc.wait(timeout=5)

            except subprocess.TimeoutExpired:

                self.proc.kill()

                self.proc.wait(timeout=5)