"""Network-fault family: the fault proxy sits between the client and a loopback server.

Oak: client -> proxy (fixture port) -> ``oak serve`` (ephemeral port).

Git: client -> proxy (fixture port) -> stdlib smart-HTTP ``git http-backend``.

A clean record run enumerates the operation's requests ("METHOD /path" +

occurrence). Each trial then injects exactly one per-request fault (or one

global slow/jitter mode) and is judged by the crash-family invariants plus:

- ``no_false_success``            exit 0 => exactly the calibrated outcome

- ``unknown_not_reported_success`` when the reply to a mutating request was

  lost (drop/truncate/kill-after-upstream) the client must not exit 0 without

  having talked to the server again (reconciled) afterwards

"""

from __future__ import annotations

import os

import signal

import subprocess

import time

from pathlib import Path

from typing import Callable, Optional

from . import FAULT_NOT_FIRED, HARNESS_ERROR

from .crash import evaluate

from .engine import Calibration, Trial, classify, converge_check

from .faultproxy import GLOBAL_KINDS, MUTATING_METHODS, Fault, FaultProxy

from .fixtures import Fixture

from .procs import Result, free_port

from .readers import worktree_manifest

PROXY_OPS = ("push", "pull_ff", "pull_diverged", "clone")

PER_REQUEST = ("fail_status", "drop_after_send", "truncate_reply", "kill_before_forward", "kill_after_upstream")

MAX_POINTS = 8

DETERMINISM_ENV = {"OAK_UPLOAD_CONCURRENCY": "1", "OAK_DOWNLOAD_CONCURRENCY": "1"}

class ProxiedTrial:

    """A Trial whose server listens behind a FaultProxy on the fixture port."""

    def __init__(self, fixture: Fixture, subject_bin: str, work: Path, name: str, faults: list[Fault]):

        self.trial = Trial(fixture, subject_bin, work, name)

        self.faults = faults

        self.pid: Optional[int] = None

        self.proxy: Optional[FaultProxy] = None

        self.gitserver = None

    def kill_client(self) -> None:

        if self.pid:

            try:

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

            except ProcessLookupError:

                pass

    def start(self) -> None:

        t = self.trial

        upstream = free_port()

        if t.kind == "oak":

            t.server.start(upstream)

        else:

            from oakbench.gitserver import GitHTTPServer

            self.gitserver = GitHTTPServer(t.dir / "srv", port=upstream, git_bin=t.adapter.bin).start()

            t.adapter.must([t.adapter.bin, "remote", "set-url", "origin", self.url + "/remote.git"], t.client)

        self.proxy = FaultProxy(upstream, self.faults, kill_hook=self.kill_client,

                                listen_port=t.fixture.port).start()

    @property

    def url(self) -> str:

        return f"http://127.0.0.1:{self.trial.fixture.port}"

    def clone_url(self) -> str:

        return self.url if self.trial.kind == "oak" else self.url + "/remote.git"

    def stop(self) -> None:

        if self.proxy:

            self.proxy.stop()

        if self.gitserver:

            self.gitserver.stop()

        self.trial.stop_server()

    def run_op(self, timeout: float = 120) -> Result:

        t = self.trial

        argv = t.op_argv(url=self.clone_url())

        # Per-request faults need a deterministic request sequence; global jitter/slow

        # modes keep the subject's default concurrency so jitter actually reorders replies.

        deterministic = not any(f.kind in GLOBAL_KINDS for f in self.faults)

        env = t.adapter.env(DETERMINISM_ENV if deterministic else None)

        start = time.perf_counter()

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

                                stdin=subprocess.DEVNULL, start_new_session=True)

        self.pid = proc.pid

        try:

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

            timed_out = False

        except subprocess.TimeoutExpired:

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

            out, err = proc.communicate()

            timed_out = True

        self.pid = None

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

                      err.decode("utf-8", "replace"), (time.perf_counter() - start) * 1000.0,

                      killed_by_harness=timed_out, timed_out=timed_out)

def record(fixture: Fixture, subject_bin: str, work: Path) -> tuple[Calibration, list[tuple[str, int]], list]:

    pt = ProxiedTrial(fixture, subject_bin, work, "proxy-record", [])

    try:

        pt.start()

        pt.trial.pre()

        res = pt.run_op()

        pt.trial.settle()

        if res.returncode != 0:

            return Calibration(res.elapsed_ms, None, None, None, None, ok=False,

                               detail=f"unfaulted proxied op failed rc={res.returncode}: {res.stderr[-300:]}"), [], []

        repo = pt.trial.target_repo()

        tree, status = pt.trial.adapter.tree(repo)

        calib = Calibration(res.elapsed_ms, tree, worktree_manifest(repo), pt.trial.adapter.head(repo),

                            pt.trial.server_head(), detail=f"tree_reader={status}")

        keys = pt.proxy.keys_in_order()

        methods = [(r.key, r.occurrence, r.method) for r in sorted(pt.proxy.records, key=lambda r: r.index)]

        calib.requests = [f"{k}#{o}" for k, o in keys]

        return calib, keys, methods

    finally:

        pt.stop()

        pt.trial.cleanup()

def fault_points(keys: list[tuple[str, int]]) -> list[tuple[str, int]]:

    """Every distinct request key's first and last occurrence, capped at MAX_POINTS."""

    last: dict[str, int] = {}

    order: list[str] = []

    for key, occ in keys:

        if key not in last:

            order.append(key)

        last[key] = max(last.get(key, 0), occ)

    points: list[tuple[str, int]] = []

    for key in order:

        points.append((key, 1))

        if last[key] > 1:

            points.append((key, last[key]))

    if len(points) > MAX_POINTS:

        # Mutating requests (unknown-outcome risk) always stay; reads fill the rest evenly.

        mutating = [p for p in points if p[0].split(" ", 1)[0] in MUTATING_METHODS]

        reads = [p for p in points if p not in mutating]

        room = max(0, MAX_POINTS - len(mutating))

        step = len(reads) / room if room else 0

        chosen = set(mutating) | {reads[int(i * step)] for i in range(room)} if reads else set(mutating)

        points = [p for p in points if p in chosen]

    return points

def run_proxy_trial(fixture: Fixture, subject_bin: str, work: Path, name: str, calib: Calibration,

                    fault: Fault, keep: Callable[[Trial, str], None]) -> dict:

    pt = ProxiedTrial(fixture, subject_bin, work, name, [fault])

    started = time.perf_counter()

    op_res: Optional[Result] = None

    try:

        pt.start()

        pt.trial.pre()

        op_res = pt.run_op()

        fired = [f for f in pt.proxy.fired()]

        # judge with the crash-family evaluator (it also runs the next op through the proxy,

        # whose one-shot fault is already consumed)

        inv, obs = evaluate(pt.trial, calib, op_res, bool(fired))

        obs["requests"] = len(pt.proxy.records)

        faulted = [r for r in pt.proxy.records if r.tag and not r.tag.startswith("SLOW") and "CLIENT_GONE" not in r.tag]

        obs["faulted_requests"] = [f"{r.key}#{r.occurrence}:{r.tag}" for r in faulted][:3]

        if op_res.returncode == 0:

            ok, detail = converge_check(pt.trial, calib, pt.trial.target_repo())

            inv["op_success_is_true"] = ok

            obs["op_converge"] = detail

            unknown = [r for r in faulted if r.method in MUTATING_METHODS

                       and r.tag in ("REPLY_DROPPED", "REPLY_TRUNCATED")]

            if unknown and fault.kind not in GLOBAL_KINDS:

                first = min(r.index for r in unknown)

                later = [r for r in pt.proxy.records if r.index > first]

                inv["unknown_not_reported_success"] = bool(later)

                obs["reconciled_after_unknown"] = len(later)

        if op_res.returncode not in (0, None) and fixture.op == "push":

            # Known-failure report for a mutation that was in fact applied: not an

            # invariant (failure is the safe direction) but worth surfacing.

            obs["applied_but_reported_failure"] = pt.trial.server_head() == calib.post_server_head

        classification = classify(inv, fault_fired=bool(fired), reader_errors=obs.get("reader_errors"))

        error = None

    except Exception as exc:  # noqa: BLE001

        inv, obs, classification, error = {}, {}, HARNESS_ERROR, f"{type(exc).__name__}: {exc}"

        fired = []

    finally:

        pt.stop()

    row = {

        "fault_point": fault.describe(),

        "fault_fired": bool(fired),

        "invariants": inv,

        "violations": sorted(k for k, v in inv.items() if v is False),

        "observations": obs,

        "classification": classification,

        "elapsed_ms": round(op_res.elapsed_ms if op_res else (time.perf_counter() - started) * 1000.0, 3),

        "command": op_res.argv if op_res else [],

    }

    if error:

        row["harness_error"] = error

    elif classification == HARNESS_ERROR and obs.get("reader_errors"):

        row["harness_error"] = "independent reader failed: " + "; ".join(obs["reader_errors"][:2])

    if classification == FAULT_NOT_FIRED:

        row["skip_reason"] = f"proxy fault {fault.describe()} never matched a request"

    keep(pt.trial, classification)

    return row

def plan(keys: list[tuple[str, int]], methods: list, seeds: tuple[int, ...] = (1, 2)) -> list[Fault]:

    faults: list[Fault] = []

    for key, occ in fault_points(keys):

        for kind in PER_REQUEST:

            faults.append(Fault(kind, key=key, occurrence=occ, status=503))

    faults.append(Fault("slow_reply", bytes_per_s=256 * 1024))

    for seed in seeds:

        faults.append(Fault("jitter", max_ms=120, seed=seed))

    return faults