"""Conformance gate: diagnostic conformance rows vs an explicit expected-failures list.

Semantics (strict xfail, schema 2):

- every expected-failure entry names its failing legs AND an error pattern per

  leg (``re.fullmatch`` against that leg's detail; legacy v1 scenarios use the

  pseudo-leg ``behaviour`` matched against the row's ``failure``). A listed

  behaviour fails as expected only when the failing-leg set is identical and

  every detail matches its pattern. Any other failure - a new leg, a changed

  error (EEXIST -> ``extra=['../../etc/evil']``, a rejection -> a panic) - is

  a REGRESSION;

- a listed behaviour that passes on a profile the entry applies to is FIXED:

  not a regression, but publish_gate refuses until the list is updated;

- entries may be scoped to filesystem profiles (``fs_profiles``); on other

  profiles they do not apply, so a pass there is just a pass (never FIXED) and

  a failure there is a regression;

- ``outcome: refused`` marks typed policy refusals (a product decision, e.g.

  control characters in names) and is reported separately from bugs;

  ``outcome: intermittent`` marks a timing-dependent bug: a pass is a pass

  (never FIXED), the pinned failure is INTERMITTENT, any other failure is a

  regression;

- coverage: for a recorded profile, any skip not in its ``expected_skips`` is

  a coverage loss (UNMEASURED) - a behaviour that used to pass or xfail must

  not silently turn into a skip. On an unrecorded profile only behaviours that

  need a known filesystem-dependent capability (case/normalization/non-UTF-8)

  or are gated may skip; a row without ``fs_profile`` and a run with zero

  measured behaviours are UNMEASURED. publish_gate additionally requires a

  recorded profile;

- harness errors, reader-unmeasured legs and required behaviours with no row

  make the result UNMEASURED;

- error patterns must be specific: the loader rejects any pattern that matches

  the empty string, matches a nonsense/generic sentinel string (``panicked``,

  ``corrupt``, random text ...) or contains no literal character at all

  (only wildcards/classes), because such a pattern would hide any error

  under the expected failure;

- ``pending:`` tickets are listed in every summary; publish_gate refuses any

  entry that still carries one (devloop only notes them).

Only the gated subject (Oak) is judged; Git rows are the reference arm.

"""

from __future__ import annotations

import json

from pathlib import Path

import random

import re

import string

try:  # Python >= 3.11

    import re._parser as _sre_parse  # type: ignore[import-not-found]

except ImportError:  # pragma: no cover - Python < 3.11

    import sre_parse as _sre_parse  # type: ignore[no-redef]

ROOT = Path(__file__).resolve().parents[2]

EXPECTED_FAILURES = ROOT / "config" / "conformance_expected_failures.json"

SCHEMA_VERSION = 2

# fb-N once filed; pending:<slug> only while the coordinator files it (listed

# in every summary so it cannot be forgotten).

TICKET = re.compile(r"^(fb-\d+|pending:[a-z0-9-]+)$")

OUTCOMES = ("bug", "refused", "intermittent")

CAPABILITY_SKIP_PREFIXES = ("filesystem capability unavailable", "gated:", "gated behaviour")

# Capabilities that legitimately differ between ordinary hosts (macOS APFS vs

# Linux ext4). On an unrecorded profile only behaviours needing one of these,

# or gated behaviours, may skip; any other skip means a broken probe or host

# and the result is UNMEASURED.

KNOWN_FS_DEPENDENT = frozenset({"case_insensitive_fs", "case_sensitive_fs", "normalization_sensitive_fs",

                                "non_utf8_names"})

# A legitimate expected-failure pattern pins one concrete error; it must match

# none of these. They cover generic failure words (so "panicked"/"corrupt"

# cannot hide under an xfail), the kind of text a catch-all pattern swallows,

# and deterministic random noise.

_rng = random.Random(0x0A4B)

SENTINELS = ("", " ", "\n", "x", "panicked", "corrupt", "error", "failed", "zzz-sentinel",

             "thread 'main' panicked at src/main.rs:1:1", "database disk image is malformed",

             "internal error", "HTTP 500 Internal Server Error", "Segmentation fault: 11",

             "zzz-sentinel: no such behaviour exited 99",

             *("".join(_rng.choice(string.ascii_letters + string.digits + string.punctuation + " ")

                       for _ in range(_rng.randint(1, 60))) for _ in range(24)))

del _rng

def _has_literal(parsed) -> bool:

    """True if the parsed regex contains a literal character outside a character class."""

    for op, arg in parsed:

        name = str(op)

        if name == "LITERAL":

            return True

        if name in ("NOT_LITERAL", "IN", "ANY", "AT", "CATEGORY", "GROUPREF", "GROUPREF_EXISTS"):

            continue

        if name == "SUBPATTERN":

            if _has_literal(arg[-1]):

                return True

        elif name in ("MAX_REPEAT", "MIN_REPEAT", "POSSESSIVE_REPEAT"):

            low, _high, sub = arg

            if low > 0 and _has_literal(sub):

                return True

        elif name == "BRANCH":

            # every alternative must carry a literal for the branch to count

            if all(_has_literal(alt) for alt in arg[1]):

                return True

        elif name == "ATOMIC_GROUP":

            if _has_literal(arg):

                return True

    return False

def pattern_too_broad(pattern: str) -> str | None:

    """Why an expected-failure error pattern is a catch-all, or None if it is specific."""

    compiled = re.compile(pattern, re.S)

    for sentinel in SENTINELS:

        if compiled.fullmatch(sentinel):

            return f"matches the sentinel {sentinel!r}"

    if not _has_literal(_sre_parse.parse(pattern, re.S)):

        return "has no required literal text (wildcards/classes only)"

    return None

def fs_dependent_skip_allowed(behaviour: str) -> bool:

    from oakbench.conformance_matrix import BY_ID

    spec = BY_ID.get(behaviour)

    return spec is not None and (spec.gated or bool(spec.requires & KNOWN_FS_DEPENDENT))

def load_expected(path: Path = EXPECTED_FAILURES) -> dict:

    data = json.loads(Path(path).read_text())

    if data.get("schema_version") != SCHEMA_VERSION or not isinstance(data.get("entries"), list):

        raise ValueError("conformance expected-failures: unsupported schema")

    entries = {}

    for entry in data["entries"]:

        behaviour = entry.get("behaviour")

        if not behaviour or behaviour in entries:

            raise ValueError(f"conformance expected-failures: bad or duplicate behaviour {behaviour!r}")

        if not TICKET.match(str(entry.get("ticket", ""))):

            raise ValueError(f"conformance expected-failures: {behaviour} needs a ticket (fb-N or pending:slug)")

        if entry.get("outcome", "bug") not in OUTCOMES:

            raise ValueError(f"conformance expected-failures: {behaviour} outcome must be one of {OUTCOMES}")

        legs = entry.get("failed_legs")

        if not isinstance(legs, dict) or not legs:

            raise ValueError(f"conformance expected-failures: {behaviour} needs failed_legs {{leg: pattern}}")

        compiled = {}

        for leg, pattern in legs.items():

            if not isinstance(pattern, str) or not pattern:

                raise ValueError(f"conformance expected-failures: {behaviour}/{leg} needs an error pattern")

            try:

                broad = pattern_too_broad(pattern)

            except re.error as exc:

                raise ValueError(f"conformance expected-failures: {behaviour}/{leg} pattern invalid: {exc}")

            if broad:

                raise ValueError(f"conformance expected-failures: {behaviour}/{leg} pattern {pattern!r} is a "

                                 f"catch-all ({broad}); pin the concrete error")

            compiled[leg] = re.compile(pattern, re.S)

        profiles = entry.get("fs_profiles")

        if profiles is not None and (not isinstance(profiles, list) or not profiles):

            raise ValueError(f"conformance expected-failures: {behaviour} fs_profiles must be a non-empty list")

        entries[behaviour] = {**entry, "outcome": entry.get("outcome", "bug"), "patterns": compiled}

    coverage = data.get("coverage") or {}

    if not isinstance(coverage, dict):

        raise ValueError("conformance expected-failures: coverage must map profile -> {expected_skips}")

    return {"subject": data.get("subject", "oak"), "entries": entries, "coverage": coverage}

def pending_tickets(expected: dict) -> list:

    """[(behaviour, ticket, outcome)] for entries whose ticket is still pending:<slug>."""

    return sorted((behaviour, entry["ticket"], entry["outcome"]) for behaviour, entry in expected["entries"].items()

                  if str(entry["ticket"]).startswith("pending:"))

def row_failed_legs(row: dict) -> dict:

    """{failed leg: detail} for a failing row (legacy rows: {'behaviour': failure})."""

    legs = row.get("legs")

    if isinstance(legs, dict):

        return {name: str(leg.get("detail") or "") for name, leg in legs.items()

                if leg.get("status") in ("fail", "error")}

    return {"behaviour": str(row.get("failure") or "")} if row.get("returncode") == 1 else {}

def entry_applies(entry: dict, profile: str | None) -> bool:

    profiles = entry.get("fs_profiles")

    return profiles is None or profile in profiles

def mismatch(entry: dict, failed: dict) -> str | None:

    """Why a failing row does not match its expected failure, or None."""

    expected = set(entry["patterns"])

    if set(failed) != expected:

        return f"expected legs {sorted(expected)}, observed {sorted(failed)}"

    for leg, pattern in entry["patterns"].items():

        if not pattern.fullmatch(failed[leg]):

            return f"leg {leg} error changed: {failed[leg][:200]!r} does not match {pattern.pattern!r}"

    return None

def evaluate(rows: list, expected: dict, required=None) -> dict:

    subject = expected["subject"]

    entries = expected["entries"]

    result = {"subject": subject, "passed": [], "expected_failures": [], "refused": [], "fixed": [],

              "intermittent": [], "regressions": [], "unmeasured": [], "skipped": [], "git_reference_failures": [],

              "differential_differs": [], "profiles": [], "notes": []}

    seen = set()

    subject_rows = [row for row in rows if row.get("evidence_class") == "diagnostic_conformance"]

    profiles = sorted({str(row.get("fs_profile")) for row in subject_rows

                       if row.get("subject") == subject and row.get("fs_profile")})

    result["profiles"] = profiles

    if len(profiles) > 1:

        result["unmeasured"].append(f"rows mix filesystem profiles {profiles}; gate one host at a time")

        result["verdict"] = "UNMEASURED"

        return result

    profile = profiles[0] if len(profiles) == 1 else None

    missing_profile = sorted(str(row.get("scenario")) for row in subject_rows

                             if row.get("subject") == subject and not row.get("fs_profile"))

    if missing_profile:

        result["unmeasured"].append(f"rows without fs_profile cannot be judged: {missing_profile[:5]}")

    recorded = expected["coverage"].get(profile) if profile else None

    result["profile_recorded"] = recorded is not None

    if profile and recorded is None:

        result["notes"].append(f"filesystem profile {profile} has no recorded coverage; only known "

                               f"filesystem-dependent ({sorted(KNOWN_FS_DEPENDENT)}) and gated skips accepted")

    expected_skips = set((recorded or {}).get("expected_skips", []))

    for row in subject_rows:

        behaviour = row.get("scenario")

        code = row.get("returncode")

        if row.get("subject") != subject:

            if row.get("subject") == "git" and code == 1:

                result["git_reference_failures"].append(f"{behaviour}: {row.get('failure')}")

            continue

        seen.add(behaviour)

        entry = entries.get(behaviour)

        if entry is not None and not entry_applies(entry, profile):

            entry = None

        differential = str(row.get("git_differential") or "")

        if differential.startswith("differs"):

            result["differential_differs"].append(f"{behaviour}: {differential}")

        if code == 0:

            if entry and entry["outcome"] == "intermittent":

                # An intermittent failure is expected to pass most runs: a pass

                # is a pass, never evidence of a fix.

                result["passed"].append(behaviour)

            elif entry:

                result["fixed"].append(f"{behaviour}: now passes; remove {entry['ticket']} from the expected-failures list")

            else:

                result["passed"].append(behaviour)

        elif code == 1:

            failed = row_failed_legs(row)

            if entry is None:

                result["regressions"].append(f"{behaviour}: {row.get('failure')}")

                continue

            why = mismatch(entry, failed)

            if why:

                result["regressions"].append(f"{behaviour}: known failure ({entry['ticket']}) changed: {why}")

            else:

                key = {"refused": "refused", "intermittent": "intermittent"}.get(entry["outcome"], "expected_failures")

                result[key].append(f"{behaviour}: {entry['ticket']}")

        elif code == 77 and str(row.get("skip_reason", "")).startswith(CAPABILITY_SKIP_PREFIXES):

            if recorded is not None and behaviour not in expected_skips:

                result["unmeasured"].append(f"{behaviour}: coverage loss on {profile}: skipped "

                                            f"({row.get('skip_reason')}) but not an expected skip")

            elif recorded is None and not fs_dependent_skip_allowed(behaviour):

                result["unmeasured"].append(f"{behaviour}: skipped on unrecorded profile {profile} "

                                            f"({row.get('skip_reason')}); only known filesystem-dependent "

                                            "or gated behaviours may skip there")

            else:

                result["skipped"].append(f"{behaviour}: {row.get('skip_reason')}")

        else:

            result["unmeasured"].append(f"{behaviour}: {row.get('failure') or row.get('skip_reason') or code}")

    for behaviour in sorted(set(required or ()) - seen):

        result["unmeasured"].append(f"{behaviour}: no {subject} conformance row")

    if not seen:

        result["unmeasured"].append(f"no {subject} conformance rows")

    measured = sum(len(result[key]) for key in ("passed", "expected_failures", "refused", "intermittent", "fixed",

                                                "regressions"))

    pending = sorted({entry["ticket"] for entry in entries.values() if entry["ticket"].startswith("pending:")})

    if pending:

        result["notes"].append(f"unfiled tickets in the expected-failures list: {pending}")

    result["measured"] = measured

    if seen and measured == 0:

        result["unmeasured"].append(f"zero measured {subject} behaviours (every row skipped or unmeasured)")

    result["verdict"] = ("REGRESSED" if result["regressions"] else

                         "UNMEASURED" if result["unmeasured"] else "PASS")

    return result

def summary_lines(result: dict) -> list:

    lines = [f"conformance ({result['subject']}, {','.join(result['profiles']) or 'no profile'}): "

             f"{result['verdict']} - {len(result['passed'])} pass, {len(result['expected_failures'])} expected-fail, "

             f"{len(result['refused'])} policy-refused, {len(result['intermittent'])} intermittent-fail, "

             f"{len(result['fixed'])} fixed, "

             f"{len(result['regressions'])} regression(s), {len(result['unmeasured'])} unmeasured, "

             f"{len(result['skipped'])} capability/gated skip(s)"]

    for key, label in (("regressions", "REGRESSION"), ("fixed", "FIXED"), ("unmeasured", "UNMEASURED"),

                       ("expected_failures", "XFAIL"), ("refused", "REFUSED"), ("intermittent", "INTERMITTENT"), ("differential_differs", "GIT-DIFF"),

                       ("git_reference_failures", "GIT-REF"), ("skipped", "SKIP"), ("notes", "NOTE")):

        lines += [f"  {label} {item}" for item in result[key]]

    return lines