"""Run scenarios: git merge-ort oracle vs Oak's merge engine, classify, apply merge laws."""

from __future__ import annotations

import concurrent.futures

import hashlib

import itertools

import json

from pathlib import Path

import queue

import shutil

import subprocess

import time

import traceback

from typing import Callable, Dict, List, Optional

from . import classify as cl

from .gitops import Git, git_conflicts, prepare_git_home

from .oak_adapter import (LocalOakServer, PATH_DESCRIPTION, PATH_EXERCISED, oak_env,

                          run_pull_sync)

SCHEMA_VERSION = 1

def file_sha256(path: str) -> Optional[str]:

    try:

        return hashlib.sha256(Path(path).read_bytes()).hexdigest()

    except OSError:

        return None

def tree_digest(tree: Optional[dict]) -> Optional[str]:

    if tree is None:

        return None

    return hashlib.sha256(json.dumps(sorted(tree.items())).encode()).hexdigest()

def tool_version(binary: str, *args: str) -> Optional[str]:

    try:

        return subprocess.run([binary, *args], capture_output=True, text=True, timeout=30).stdout.strip() or None

    except OSError:

        return None

def git_oracle(git: Git, direction: dict) -> dict:

    merge = git.merge_tree(direction["oracle"]["ours"], direction["oracle"]["theirs"])

    return {"clean": merge["clean"], "tree": merge["tree"], "conflicts": git_conflicts(merge),

            "messages": [m for m in merge["messages"] if m["type"].startswith("CONFLICT")][:20]}

def text_evidence(git: Git, trees: dict, oak_tree: dict, oak_conf: set, case_dir: Path,

                  base: dict, ours: dict, theirs: dict) -> Dict[str, dict]:

    """Bytes for paths both sides changed that Oak text-merged (needed by the no-lost-change law)."""

    out = {}

    for path, value in oak_tree.items():

        if path in oak_conf:

            continue

        b, o, t = base.get(path), ours.get(path), theirs.get(path)

        if o is None or t is None or o == t or o == b or t == b or value in (b, o, t):

            continue

        def blob(commit):

            p = git.run("cat-file", "-p", f"{commit}:{path}", check=False)

            return p.stdout if p.returncode == 0 else b""

        merged_path = case_dir / "ours" / path

        out[path] = {"base": blob(trees["base"]) if b else b"", "ours": blob(trees["ours"]),

                     "theirs": blob(trees["theirs"]),

                     "merged": merged_path.read_bytes() if merged_path.is_file() else b""}

    return out

class Suite:

    def __init__(self, workdir: Path, oak_bin: str, git_bin: str, jobs: int = 4,

                 keep_equal: bool = False, run_id: Optional[str] = None):

        self.workdir, self.oak_bin, self.git_bin = workdir, oak_bin, git_bin

        self.jobs, self.keep_equal = max(1, jobs), keep_equal

        self.run_id = run_id or time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())

        self.git_home = workdir / "git-home"

        prepare_git_home(self.git_home, git_bin)

        self.homes: "queue.Queue[Path]" = queue.Queue()

        for i in range(self.jobs):

            home = workdir / "oak-homes" / f"h{i}"

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

            prepare_git_home(home, git_bin)

            self.homes.put(home)

        self.meta = {

            "oak": {"bin": oak_bin, "sha256": file_sha256(shutil.which(oak_bin) or oak_bin),

                    "version": tool_version(oak_bin, "--version"),

                    "path_exercised": PATH_EXERCISED, "path_description": PATH_DESCRIPTION},

            "git": {"bin": git_bin, "sha256": file_sha256(git_bin), "version": tool_version(git_bin, "--version"),

                    "role": "reference oracle (merge-ort via merge-tree --write-tree) and fixture builder"},

        }

        self._keys = itertools.count(1)

    def run(self, scenarios: List[dict], emit: Callable[[dict], None]) -> List[dict]:

        server_env = oak_env(self.workdir / "oak-homes" / "server", self.git_bin)

        (self.workdir / "oak-homes" / "server").mkdir(parents=True, exist_ok=True)

        server = LocalOakServer(self.workdir, self.oak_bin, server_env)

        rows: List[dict] = []

        try:

            with concurrent.futures.ThreadPoolExecutor(self.jobs) as pool:

                futures = [pool.submit(self._scenario, s, server) for s in scenarios]

                for fut in concurrent.futures.as_completed(futures):

                    for row in fut.result():

                        rows.append(row)

                        emit(row)

        finally:

            server.close()

        return rows

    def _next_key(self) -> str:

        return f"suite/m{next(self._keys):05d}"

    def _scenario(self, scenario: dict, server: LocalOakServer) -> List[dict]:

        home = self.homes.get()

        try:

            return self._scenario_rows(scenario, server, home)

        finally:

            self.homes.put(home)

    def _scenario_rows(self, scenario: dict, server: LocalOakServer, home: Path) -> List[dict]:

        git = Git(scenario["repo"], self.git_bin, self.git_home)

        safe = scenario["id"].replace("/", "__")

        scen_dir = self.workdir / "cases" / safe

        rows, results = [], {}

        bare = scen_dir / "src.git"

        for direction in scenario["directions"]:

            start = time.monotonic()

            case_dir = scen_dir / direction["name"]

            row = self._base_row(scenario, direction)

            oak_result, git_result, harness = None, None, None

            base, ours, theirs = None, {}, {}

            try:

                if scenario.get("unsupported") is None:

                    git_result = git_oracle(git, direction)

                    trees = direction["trees"]

                    unique_base = trees.get("base") and len(direction.get("merge_bases") or [1]) == 1

                    base = git.tree(trees["base"]) if unique_base else None

                    ours, theirs = git.tree(trees["ours"]), git.tree(trees["theirs"])

                    if not bare.exists():

                        self._make_bare(git, bare, scenario)

                    if case_dir.exists():

                        shutil.rmtree(case_dir)

                    case_dir.mkdir(parents=True)

                    env = oak_env(home, self.git_bin)

                    oak_result = run_pull_sync(server, self.oak_bin, env, Git(bare, self.git_bin, self.git_home),

                                               bare, case_dir, self._next_key(), direction["plan"])

            except Exception as exc:  # instrument failure, never an Oak verdict

                harness = f"{type(exc).__name__}: {exc}"

                row["traceback"] = traceback.format_exc()[-2000:]

            texts = None

            if oak_result and oak_result.get("tree") is not None and git_result:

                texts = text_evidence(git, direction["trees"], oak_result["tree"],

                                      set(oak_result.get("conflicts") or {}), case_dir, base or {}, ours, theirs)

            verdict = cl.classify(git_result, oak_result, base, ours, theirs,

                                  unsupported=scenario.get("unsupported"), harness_error=harness,

                                  expect=direction.get("expect"), texts=texts)

            row.update(self._result_fields(git_result, oak_result, verdict))

            if direction.get("intent_note"):

                row["intent_note"] = direction["intent_note"]

            row["elapsed_ms"] = round((time.monotonic() - start) * 1000)

            row["case_dir"] = str(case_dir) if case_dir.exists() else None

            if verdict["class"] == "equal-to-git" and not self.keep_equal and case_dir.exists():

                shutil.rmtree(case_dir, ignore_errors=True)

                row["case_dir"] = None

            results[direction["name"]] = oak_result

            rows.append(row)

        if len(rows) == 2:

            sym = cl.symmetry(*(results[d["name"]] for d in scenario["directions"]))

            for row in rows:

                row["laws"]["symmetric"] = None if sym is None else sym["holds"]

                if sym and not sym["holds"]:

                    row.setdefault("evidence", {})["symmetry"] = sym

        if not self.keep_equal and all(r["class"] == "equal-to-git" for r in rows):

            shutil.rmtree(scen_dir, ignore_errors=True)

        return rows

    def _make_bare(self, git: Git, bare: Path, scenario: dict) -> None:

        """Bare mirror holding every plan commit under a ref (builder commits are unreferenced;

        `oak clone <x.git>` imports from the mirror's HEAD, which the adapter points per step)."""

        wanted = set()

        for d in scenario["directions"]:

            plan = d["plan"]

            wanted.update(x for x in (plan.get("main"), plan.get("fork"), plan.get("final_stage")) if x)

            for step in plan["steps"]:

                wanted.update(x for x in (step.get("tree"), step.get("stage")) if x)

        ns = "refs/mergesuite/" + hashlib.sha256(scenario["id"].encode()).hexdigest()[:16]

        for i, commit in enumerate(sorted(wanted)):

            git.run("update-ref", f"{ns}/keep{i}", commit)

        bare.parent.mkdir(parents=True, exist_ok=True)

        Git(bare.parent, self.git_bin, self.git_home).run("init", "-q", "--bare", str(bare))

        Git(bare, self.git_bin, self.git_home).run("fetch", "-q", str(scenario["repo"]),

                                                   f"+{ns}/*:refs/mergesuite/*")

    def _base_row(self, scenario: dict, direction: dict) -> dict:

        return {

            "merge_suite_schema_version": SCHEMA_VERSION,

            "evidence_class": "diagnostic_merge_correctness",

            "run_id": self.run_id,

            "scenario": scenario["id"], "scenario_version": scenario["version"],

            "direction": direction["name"], "origin": scenario["origin"], "upstream": scenario["upstream"],

            "notes": scenario.get("notes") or None,

            "oak": self.meta["oak"], "git": self.meta["git"],

            "laws": {"no_lost_change": None, "idempotent": None, "symmetric": None,

                     "false_conflict": None, "missed_conflict": None, "prediction_agrees": None},

            "repro": ("python3 suites/merge/run_merge_suite.py --workdir <dir> --only "

                      f"'{scenario['id']}'"),

        }

    def _result_fields(self, git_result: Optional[dict], oak_result: Optional[dict], verdict: dict) -> dict:

        out = {"class": verdict["class"], "reasons": verdict.get("reasons", [])}

        if verdict.get("notes"):

            out["class_notes"] = verdict["notes"]

        evidence = dict(verdict.get("evidence") or {})

        laws = {"no_lost_change": None, "idempotent": None, "symmetric": None,

                "false_conflict": None, "missed_conflict": None, "prediction_agrees": None}

        if git_result:

            out["git_result"] = {"clean": git_result["clean"], "conflicts": git_result["conflicts"],

                                 "tree_digest": tree_digest(git_result["tree"]),

                                 "conflict_messages": git_result["messages"]}

        if oak_result:

            out["oak_result"] = {

                "path_exercised": oak_result.get("path_exercised"),

                "clean": oak_result.get("clean"), "conflicts": oak_result.get("conflicts"),

                "tree_digest": tree_digest(oak_result.get("tree")),

                "pull_returncode": oak_result.get("pull_returncode"),

                "pull_receipt": oak_result.get("pull_receipt"),

                "prediction": oak_result.get("prediction"),

                "idempotence": oak_result.get("idempotence"),

                "worktree_matches_head": oak_result.get("worktree_matches_head"),

                "result_source": oak_result.get("result_source"),

                "conflict_worktree_stale_paths": oak_result.get("conflict_worktree_stale_paths") or None,

                "error": oak_result.get("error"), "error_kind": oak_result.get("error_kind"),

                "shim_log": oak_result.get("shim_log") or None,

            }

            if verdict["class"] not in ("equal-to-git",):

                out["oak_steps"] = [{k: s[k] for k in ("label", "argv", "returncode", "ms")}

                                    | {"stderr_tail": s["stderr"][-300:]} for s in oak_result.get("steps", [])]

            if oak_result.get("clean") is not None and git_result:

                law = evidence.get("law_no_lost_change")

                laws["no_lost_change"] = None if law is None else not (law.get("lost") or law.get("fabricated"))

                laws["idempotent"] = cl.idempotence_holds(oak_result)

                oak_conf = set(oak_result.get("conflicts") or {})

                git_conf = set(git_result["conflicts"])

                laws["false_conflict"] = bool(oak_conf - git_conf)

                laws["missed_conflict"] = bool(git_conf - oak_conf)

                pred = oak_result.get("prediction") or {}

                if pred.get("available"):

                    laws["prediction_agrees"] = (bool(pred.get("clean")) == bool(oak_result["clean"])

                                                 and set(pred.get("conflict_files") or []) == oak_conf)

        out["laws"] = laws

        if evidence:

            out["evidence"] = evidence

        return out

def summarize(rows: List[dict]) -> dict:

    counts = {c: 0 for c in cl.CLASSES}

    by_origin: Dict[str, Dict[str, int]] = {}

    for r in rows:

        counts[r["class"]] += 1

        by_origin.setdefault(r["origin"], {c: 0 for c in cl.CLASSES})[r["class"]] += 1

    def rate(key):

        vals = [r["laws"][key] for r in rows if r["laws"].get(key) is not None]

        return {"true": sum(1 for v in vals if v), "measured": len(vals)}

    return {"cases": len(rows), "scenarios": len({r["scenario"] for r in rows}), "counts": counts,

            "by_origin": by_origin,

            "laws": {k: rate(k) for k in ("no_lost_change", "idempotent", "symmetric", "false_conflict",

                                          "missed_conflict", "prediction_agrees")}}