"""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

import os

import re

import fnmatch

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)

# Row/reason schema. Bump whenever the reasons a row can carry change, so a

# resummarize of an older run (whose rows could not carry the new reasons) is

# detected instead of reporting known bugs as "NOT reproduced (fixed?)".

#   1: initial (P3 landing, main 5dd811a1)

#   2: an expectation-violated row also carries its law reasons; rows without a

#      unique merge base carry the base-free law checks (law_base_free)

SCHEMA_VERSION = 2

def stale_schema_rows(rows: List[dict]) -> Dict[str, int]:

    """{version: count} of rows not written under the current SCHEMA_VERSION ('missing' when absent)."""

    out: Dict[str, int] = {}

    for r in rows:

        version = r.get("merge_suite_schema_version")

        if version != SCHEMA_VERSION:

            key = "missing" if version is None else str(version)

            out[key] = out.get(key, 0) + 1

    return out

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

        if merged_path.is_symlink():

            merged = os.fsencode(os.readlink(merged_path))  # symlink "content" is its target

        else:

            merged = merged_path.read_bytes() if merged_path.is_file() else b""

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

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

    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)

        self.bugs = load_known_bugs()

    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))

            row["known_bug"] = match_known_bug(row, self.bugs)

            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.get("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 any(

                    law.get(k) for k in ("lost", "mode_lost", "markers", "markers_no_final_newline",

                                         "markers_symlink", "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)

            elif oak_result.get("error_kind") == "oak":

                # The real pull failed: a prediction that promised a merge (clean or a conflict

                # set) disagrees with reality; count it rather than dropping the row.

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

                if pred.get("available"):

                    laws["prediction_agrees"] = False

                pf = oak_result.get("post_failure") or {}

                if pf.get("hazard_head_moved_worktree_stale") or pf.get("hazard_worktree_dirty_head_unmoved"):

                    out["post_failure_hazard"] = True

            if oak_result.get("post_failure"):

                out["oak_result"]["post_failure"] = oak_result["post_failure"]

        out["laws"] = laws

        cl.validate_class(out["class"])

        if evidence:

            out["evidence"] = evidence

        return out

KNOWN_BUGS = Path(__file__).resolve().parents[1] / "known_bugs.json"

def load_known_bugs(path: Path = KNOWN_BUGS) -> List[dict]:

    return json.loads(path.read_text())["bugs"] if path.exists() else []

def row_id(row: dict) -> str:

    return f"{row['scenario']}|{row['direction']}"

def _bug_covers(bug: dict, reason: str) -> bool:

    return any(re.search(p, reason) for p in bug.get("reason_patterns") or [])

def match_known_bug(row: dict, bugs: List[dict]) -> Optional[str]:

    """The known bug that explains an oak-wrong row, or None.

    A row maps to a bug only when EVERY reason on the row is covered by that one

    bug's reason patterns (and its optional scenario glob matches). A row with an

    extra, unexplained reason stays unmapped, so a new bug cannot hide behind a

    known one."""

    if row["class"] != "oak-wrong" or not row.get("reasons"):

        return None

    for bug in bugs:

        glob = bug.get("scenario_glob")

        if glob and not fnmatch.fnmatch(row["scenario"], glob):

            continue

        if all(_bug_covers(bug, reason) for reason in row["reasons"]):

            return bug["id"]

    return None

def missed_conflict_kind(row: dict) -> str:

    kinds = set()

    for types in ((row.get("git_result") or {}).get("conflicts") or {}).values():

        kinds.update(types)

    if any("rename" in k for k in kinds):

        return "rename-or-directory-rename"

    return "/".join(sorted(kinds)) or "unknown"

def summarize(rows: List[dict], bugs: Optional[List[dict]] = None) -> dict:

    bugs = load_known_bugs() if bugs is None else bugs

    stale = stale_schema_rows(rows)

    mapped: Dict[str, List[str]] = {b["id"]: [] for b in bugs}

    unmapped: List[str] = []

    for r in rows:

        bug = match_known_bug(r, bugs)

        r["known_bug"] = bug

        if bug:

            mapped[bug].append(row_id(r))

        elif r["class"] == "oak-wrong":

            unmapped.append(row_id(r))

    in_run = {row_id(r) for r in rows}

    # A pinned row that is now a harness error (or unsupported) was not judged:

    # its bug is neither reproduced nor fixed there.

    unjudged = {row_id(r) for r in rows if r["class"] in ("harness-error", "oak-unsupported")}

    def bug_report(bug: dict) -> dict:

        # Only pinned rows this run actually executed can vanish: a subset run

        # (--groups/--only) must not report a bug whose rows it never ran as fixed.

        pinned = set(bug.get("expected_rows") or [])

        got, want = set(mapped[bug["id"]]), pinned & in_run

        new = sorted(got - pinned)

        indeterminate = sorted((want - got) & unjudged)

        vanished = sorted(want - got - unjudged)

        if stale:

            # Rows written under another reason schema cannot be mapped reliably:

            # a missing bug may just be a renamed/extended reason, not a fix.

            status = f"indeterminate (rows not on reason schema v{SCHEMA_VERSION}: {stale})"

        elif indeterminate and not vanished and not new:

            status = "harness-error (indeterminate)"

        elif not want and not got:

            status = "not in this run (no pinned row executed)"

        elif not got:

            status = "NOT reproduced (fixed?)"

        elif vanished or new:

            status = "changed"

        else:

            status = "reproduced as pinned"

        return {"family": bug.get("family"), "severity": bug.get("severity"), "rows": len(got),

                "expected_rows": len(want), "expected_rows_total": len(pinned), "new_rows": new,

                "vanished_rows": vanished, "indeterminate_rows": indeterminate, "status": status}

    missed_by_class: Dict[str, int] = {}

    missed_by_kind: Dict[str, int] = {}

    for r in rows:

        if r["laws"].get("missed_conflict"):

            missed_by_class[r["class"]] = missed_by_class.get(r["class"], 0) + 1

            k = missed_conflict_kind(r)

            missed_by_kind[k] = missed_by_kind.get(k, 0) + 1

    base = _summarize(rows)

    base.update({

        "known_bugs": {b["id"]: bug_report(b) for b in bugs},

        "oak_wrong_unmapped": len(unmapped),

        "oak_wrong_unmapped_rows": unmapped,

        "missed_conflict_by_class": missed_by_class,

        "missed_conflict_by_git_kind": missed_by_kind,

        "post_failure_hazard_rows": sum(1 for r in rows if r.get("post_failure_hazard")),

        "schema_version": SCHEMA_VERSION,

        "stale_schema_rows": stale,

    })

    return base

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")}}