"""Crash-point family: kill -9 the client at time offsets and at every durability boundary."""

from __future__ import annotations

import time

from pathlib import Path

from typing import Callable, Optional

from . import FAULT_NOT_FIRED, HARNESS_ERROR

from .crashpoints import CrashInstrument, boundary_kinds, count_boundaries

from .engine import Calibration, Trial, actionable, classify, converge_check, torn_verdict, worktree_verdict

from .fixtures import Fixture

from .procs import run, run_killed_after

from .readers import worktree_manifest

TIME_FRACTIONS = (0.02, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.97)

MAX_SYSCALL_POINTS = 16

def calibrate(fixture: Fixture, subject_bin: str, work: Path, instrument: Optional[CrashInstrument]) -> Calibration:

    trial = Trial(fixture, subject_bin, work, "calibrate")

    try:

        trial.start_server()

        trial.pre()

        res = trial.adapter.run(trial.op_argv(), trial.op_cwd())

        trial.settle()

        if res.returncode != 0:

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

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

        repo = trial.target_repo()

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

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

                            trial.server_head() if fixture.op in ("push", "clone", "pull_ff", "pull_diverged") else None,

                            detail=f"tree_reader={status}")

    finally:

        trial.cleanup()

    if instrument is not None:

        bin_copy = instrument.oak_bin if fixture.kind == "oak" else instrument.git_bin

        if bin_copy is not None:

            trial = Trial(fixture, subject_bin, work, "calibrate-syscalls")

            try:

                trial.start_server()

                trial.pre()

                log = trial.dir / "boundaries.log"

                res = trial.adapter.run(trial.op_argv(bin_override=str(bin_copy)), trial.op_cwd(),

                                        instrument.env(0, log, fixture.kind))

                calib.boundaries = count_boundaries(log) if res.returncode == 0 else 0

                calib.boundary_kinds = boundary_kinds(log)

            finally:

                trial.cleanup()

    return calib

def evaluate(trial: Trial, calib: Calibration, op_res, fault_fired: bool) -> tuple[dict, dict]:

    fx = trial.fixture

    inv: dict[str, Optional[bool]] = {}

    obs: dict = {"op": op_res.brief(400), "fault_fired": fault_fired}

    reader_errors: list[str] = []

    inv["op_no_hang"] = not op_res.timed_out

    trial.settle()

    repo = trial.target_repo()

    if fx.op == "clone":

        exists = (repo / (".oak" if trial.kind == "oak" else ".git")).exists()

        obs["clone_dest"] = "absent" if not repo.exists() else ("repo" if exists else "partial_without_store")

        inv["repo_opens"] = None

        inv["worktree_consistent"] = None

        if exists and trial.repo_opens(repo):

            inv["repo_opens"] = True

            ok, detail = worktree_verdict(trial, repo, [calib.post_worktree])

            inv["worktree_consistent"] = ok

            obs["worktree"] = detail

            ok, detail = torn_verdict(trial, repo, {}, calib.post_worktree, None, op_res)

            inv["worktree_not_torn"] = ok

            obs["torn"] = detail

        elif repo.exists():

            # A torn clone is fine only if the subject recognises it: its status must

            # refuse with an actionable error (e.g. "remove this directory and re-clone").

            dirty, st = trial.adapter.status_dirty(repo)

            obs["partial_clone_status"] = st.brief(300)

            inv["partial_clone_detected"] = dirty is None and actionable(st)

    else:

        inv["repo_opens"] = trial.repo_opens(repo)

        ok, lost = trial.acked_reachable(repo, fx.meta.get("pre_local_heads", []))

        inv["acked_local_reachable"] = ok

        if ok is None and lost:

            reader_errors += lost

        elif lost:

            obs["lost_local_heads"] = lost

        pre_wt = worktree_manifest(trial.fixture.pristine / "client")

        ok, detail = worktree_verdict(trial, repo, [pre_wt, calib.post_worktree])

        inv["worktree_consistent"] = ok

        obs["worktree"] = detail

        ok, detail = torn_verdict(trial, repo, pre_wt, calib.post_worktree, fx.meta.get("pre_head"), op_res)

        inv["worktree_not_torn"] = ok

        obs["torn"] = detail

    if trial.server is not None or trial.kind == "git":

        intended = calib.post_server_head if fx.op == "push" else None

        ok, detail = trial.server_intact(intended)

        inv["server_intact"] = ok

        obs["server"] = detail

        if "reader_error" in detail:

            reader_errors.append("server: " + detail["reader_error"])

    # next operation (no fault)

    if fx.op == "clone":

        nxt = trial.adapter.run(trial.op_argv(dest="cloned2"), trial.op_cwd())

        next_repo = trial.dir / "cloned2"

        if (trial.dir / "cloned").exists() and inv.get("repo_opens"):

            st_dirty, st_res = trial.adapter.status_dirty(trial.dir / "cloned")

            obs["crashed_clone_status"] = {"dirty": st_dirty, "rc": st_res.returncode}

    else:

        nxt = None

        for argv in trial.adapter.pre_argv(fx.op):

            staged = trial.adapter.run(argv, trial.client)

            if staged.returncode != 0:

                nxt = staged  # e.g. a stale index.lock blocks staging: judged like the op itself

                break

        if nxt is None:

            nxt = trial.adapter.run(trial.op_argv(), trial.op_cwd())

        next_repo = repo

    obs["next_op"] = nxt.brief(400)

    if nxt.timed_out:

        inv["next_op_ok_or_actionable"] = False

    elif nxt.returncode == 0:

        inv["next_op_ok_or_actionable"] = True

        trial.settle()

        ok, detail = converge_check(trial, calib, next_repo)

        inv["no_false_success"] = ok

        obs["converge"] = detail

    else:

        inv["next_op_ok_or_actionable"] = actionable(nxt)

        obs["recovery_required"] = True

    if fx.op != "clone":

        ok, lost = trial.acked_reachable(repo, fx.meta.get("pre_local_heads", []))

        inv["acked_after_next"] = ok

        if ok is None and lost:

            reader_errors += lost

        elif lost:

            obs["lost_after_next"] = lost

    if reader_errors:

        obs["reader_errors"] = reader_errors[:5]

    return inv, obs

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

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

    """fault(trial) -> (op_result, fault_fired, fault_point). Returns a partial row."""

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

    started = time.perf_counter()

    try:

        trial.start_server()

        trial.pre()

        op_res, fired, point = fault(trial)

        inv, obs = evaluate(trial, calib, op_res, fired)

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

        error = None

    except Exception as exc:  # noqa: BLE001 - recorded as harness_error, never hidden

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

        fired = None

    finally:

        trial.stop_server()

    row = {

        "fault_point": point,

        "fault_fired": 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"] = not_fired_reason(op_res, point)

    keep(trial, classification)

    return row

def not_fired_reason(op_res, point: str) -> str:

    if op_res is None:

        return f"fault at {point} did not fire"

    return (f"fault at {point} did not fire: the operation exited rc={op_res.returncode} after "

            f"{op_res.elapsed_ms:.1f}ms without being interrupted")

def time_fault(delay_s: float, argv_fn=None):

    def fault(trial: Trial):

        res = run_killed_after(trial.op_argv(), trial.op_cwd(), trial.adapter.env(), delay_s)

        # Fired only with evidence: the client died of our SIGKILL. A kill sent to an

        # already-exited (zombie) client leaves its real exit status.

        return res, res.returncode == -9 and not res.timed_out, f"t={delay_s * 1000:.1f}ms"

    return fault

TIME_RETRIES = 2

def run_time_point(fixture: Fixture, subject_bin: str, work: Path, name: str, calib: Calibration, frac: float,

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

    """Kill at ``frac`` of the op's duration; if the op beat the kill, re-aim from its observed duration.

    Calibration and trial durations differ run to run, so a late fraction can

    land after the op exited. Each retry uses a fresh copy of the fixture and

    the duration the previous attempt actually took; if the kill still never

    lands, the row is ``fault_not_fired`` (never ``held``).

    """

    delay = max(0.002, calib.elapsed_ms / 1000.0 * frac)

    attempts = []

    row: dict = {}

    for attempt in range(TIME_RETRIES + 1):

        row = run_trial(fixture, subject_bin, work, f"{name}-a{attempt}", calib, time_fault(delay), keep)

        attempts.append({"delay_ms": round(delay * 1000, 1), "fired": row.get("fault_fired"),

                         "op_elapsed_ms": row.get("elapsed_ms")})

        if row.get("fault_fired") is not False:

            break

        observed = float(row.get("elapsed_ms") or 0.0) / 1000.0

        delay = max(0.002, min(delay, observed) * frac * (0.9 ** attempt))

    row.setdefault("observations", {})["time_attempts"] = attempts

    row["observations"]["target_fraction"] = frac

    return row

def syscall_fault(instrument: CrashInstrument, n: int, kinds: list[str]):

    def fault(trial: Trial):

        bin_copy = instrument.oak_bin if trial.kind == "oak" else instrument.git_bin

        log = trial.dir / "boundaries.log"

        res = trial.adapter.run(trial.op_argv(bin_override=str(bin_copy)), trial.op_cwd(),

                                instrument.env(n, log, trial.kind))

        fired = res.returncode == -9

        kind = kinds[n - 1] if 0 < n <= len(kinds) else "?"

        return res, fired, f"before {kind}#{n}"

    return fault

def syscall_points(boundaries: int) -> list[int]:

    if boundaries <= MAX_SYSCALL_POINTS:

        return list(range(1, boundaries + 1))

    step = boundaries / MAX_SYSCALL_POINTS

    points = sorted({max(1, int(round(i * step))) for i in range(1, MAX_SYSCALL_POINTS + 1)})

    return points