#!/usr/bin/env python3
"""Validation harness for the production ADI engine (family-v1).

Checks, in order:
  1.  cross_process_reproducibility — runs compute_adi.py twice as separate
      processes and compares sha256 of both written artifacts.
  2.  engine_gates — every gate recorded in the written output passed.
  3.  seam_independent — recomputes the seam statistic from the published
      rows and matches it to the engine's recorded numbers.
  4.  gfc_independent — recomputes the worst 8-quarter window from the rows.
  5.  orientation — re-runs the perturbation gate in-process (sign read from
      each JSON's direction field), and asserts savings_rate is the only
      lower_is_worse input.
  6.  weights — registry-derived, uniform, sum to 1, zero hand-typed values.
  7.  hazen_parity — fixed tie-containing corpus through this engine's
      transform AND the sibling SDI / CDI implementations; all must agree.
      Until the SDI/CDI production ports land, the siblings are the
      committed prototypes under drafts/index_rebuild_2026-06-09/; the
      family parity test supersedes this check when those ports land.
  8.  coverage — published range contiguous; members_present bounds; the
      trailing unpublished quarters and which domains block them.

The prototype-era analyses (old-engine overlay, rank-mean degeneracy, state
panel cross-check) are recorded in the 2026-06-09 cold reviews and in the
output's methodology block; they are point-in-time audit artifacts, not
recurring validation, and do not run here.

Run from the repo root:
    PYTHONPATH=. python3 scripts/indexes/validate_adi.py

Writes data/audit/index_family/adi_validation.json.
Exit code 0 only when every check passes.
"""
from __future__ import annotations

import hashlib
import importlib.util
import json
import logging
import math
import os
import subprocess
import sys
import types
from pathlib import Path
from typing import Dict, List, Sequence, Tuple

THIS_DIR = Path(__file__).resolve().parent
REPO_ROOT = THIS_DIR.parents[1]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from scripts.indexes.family_normalization import hazen_percentiles_series  # noqa: E402
from scripts.indexes.index_output_root import index_output_root  # noqa: E402
from scripts.indexes import compute_adi  # noqa: E402

logger = logging.getLogger("validate_adi")

# The artifacts this validator grades are the ones the engine writes, so both
# follow the same root — see scripts/indexes/index_output_root.py. Grading a
# committed copy against a redirected recompute would compare two different
# runs.
OUTPUT_ROOT = index_output_root(REPO_ROOT)
OUTPUT_JSON = OUTPUT_ROOT / "data" / "indexes" / "adi.json"
OUTPUT_CSV = OUTPUT_ROOT / "data" / "indexes" / "adi.csv"
REPORT_PATH = OUTPUT_ROOT / "data" / "audit" / "index_family" / "adi_validation.json"
SIBLING_PROTOTYPES = REPO_ROOT / "drafts" / "index_rebuild_2026-06-09"


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def q_index(q: str) -> int:
    year, qq = q.split("-Q")
    return int(year) * 4 + int(qq) - 1


# ---------------------------------------------------------------------------


def check_cross_process_reproducibility() -> Dict[str, object]:
    env = {**os.environ, "PYTHONPATH": "."}
    hashes: List[Dict[str, str]] = []
    for run in (1, 2):
        proc = subprocess.run(
            [sys.executable, str(THIS_DIR / "compute_adi.py")],
            cwd=REPO_ROOT, env=env, capture_output=True, text=True,
        )
        if proc.returncode != 0:
            return {"pass": False, "error": f"run {run} exited {proc.returncode}", "stderr": proc.stderr[-2000:]}
        hashes.append({p.name: sha256(p) for p in (OUTPUT_JSON, OUTPUT_CSV)})
    return {
        "pass": hashes[0] == hashes[1],
        "run1": hashes[0],
        "run2": hashes[1],
        "rule": "two separate-process runs write byte-identical artifacts",
    }


def check_engine_gates(output: Dict) -> Dict[str, object]:
    gates = {k: bool(v["pass"]) for k, v in output["validation"].items()}
    return {"pass": all(gates.values()), "gates": gates}


def check_seam_independent(output: Dict) -> Dict[str, object]:
    rows = output["data"]
    comp = {r["quarter"]: r["composite"] for r in rows}
    qs = [r["quarter"] for r in rows]
    deltas = [abs(comp[b] - comp[a]) for a, b in zip(qs, qs[1:]) if q_index(b) - q_index(a) == 1]
    threshold = compute_adi.quantile_linear(deltas, 0.95)
    seam = abs(comp["2015-Q1"] - comp["2014-Q4"])
    engine = output["validation"]["seam"]
    agree = (abs(seam - engine["seam_abs_delta_2014Q4_to_2015Q1"]) < 0.05
             and abs(threshold - engine["threshold_p95_of_all_abs_deltas"]) < 0.05)
    return {
        "pass": seam <= threshold and agree,
        "seam_from_published_rows": round(seam, 4),
        "threshold_from_published_rows": round(threshold, 4),
        "matches_engine_numbers_within_rounding": agree,
    }


def check_gfc_independent(output: Dict) -> Dict[str, object]:
    rows = output["data"]
    comp = {r["quarter"]: r["composite"] for r in rows}
    qs = [r["quarter"] for r in rows]
    window = 8
    best_i, best_mean = 0, -1.0
    for i in range(len(qs) - window + 1):
        m = sum(comp[q] for q in qs[i:i + window]) / window
        if m > best_mean:
            best_mean, best_i = m, i
    start = qs[best_i]
    in_gfc = q_index("2008-Q1") <= q_index(start) <= q_index("2010-Q4")
    return {
        "pass": in_gfc,
        "worst_8q_window": f"{start} to {qs[best_i + window - 1]}",
        "worst_8q_window_mean": round(best_mean, 2),
        "matches_engine_window": f"{start} to {qs[best_i + window - 1]}" == output["validation"]["gfc"]["worst_8q_window"],
    }


def check_orientation() -> Dict[str, object]:
    loaded = compute_adi.load_inputs(REPO_ROOT)
    inversions = [iid for iid, li in loaded.items() if li.direction == "lower_is_worse"]
    # Assert the COMPLEMENT, not just the lower_is_worse count. Listing only one
    # value meant a member carrying a poleless direction was neither an inversion
    # nor a violation, so `only_savings` still held and the gate passed on a
    # member that cannot be oriented at all.
    unorientable = sorted(
        iid for iid, li in loaded.items()
        if li.direction not in compute_adi.ORIENTABLE_DIRECTIONS
    )
    baseline = compute_adi.build(loaded)
    gate = compute_adi.gate_orientation(loaded, baseline)
    only_savings = inversions == ["savings_rate"]
    return {
        "pass": bool(gate["pass"]) and only_savings and not unorientable,
        "inverted_inputs": inversions,
        "only_inversion_is_savings_rate": only_savings,
        "unorientable_members": unorientable,
        "every_member_carries_a_pole": not unorientable,
        "per_input": gate["per_input"],
    }


def check_weights(output: Dict) -> Dict[str, object]:
    meth = output["methodology"]
    n_domains = len(meth["domains"])
    ok = abs(meth["domain_weight"] - 1.0 / n_domains) < 1e-6
    member_detail = {}
    total = 0.0
    for d, members in meth["domains"].items():
        w_m = 1.0 / len(members)
        member_detail[d] = {"n_members": len(members), "member_weight": round(w_m, 6)}
        total += meth["domain_weight"] * w_m * len(members)
    ok = ok and abs(total - 1.0) < 1e-6
    return {"pass": ok, "domain_weight": meth["domain_weight"], "members": member_detail,
            "total_weight": round(total, 12)}


def _import_sibling(name: str, path: Path) -> types.ModuleType:
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod  # dataclass processing requires the module registered
    spec.loader.exec_module(mod)
    return mod


def check_hazen_parity() -> Dict[str, object]:
    corpora = {
        "ties_mixed": [3.2, 1.1, 3.2, 0.5, 9.9, 3.2, 1.1, 7.0],
        "all_equal": [2.0, 2.0, 2.0],
        "single": [7.7],
        "negatives": [-4.0, 0.0, -4.0, 12.5],
    }
    expected = {
        "ties_mixed": [56.25, 25.0, 56.25, 6.25, 93.75, 56.25, 25.0, 81.25],
        "all_equal": [50.0, 50.0, 50.0],
        "single": [50.0],
        "negatives": [25.0, 62.5, 25.0, 87.5],
    }
    detail: Dict[str, object] = {}
    ok = True
    for name, corpus in corpora.items():
        mine = hazen_percentiles_series(corpus)
        match = all(abs(a - b) < 1e-9 for a, b in zip(mine, expected[name]))
        detail[f"adi_{name}"] = {"values": [round(v, 4) for v in mine], "matches_expected": match}
        ok = ok and match
    # Sibling engines
    for sib, fname, adapter in (
        ("sdi", "compute_sdi.py", "dict"),
        ("cdi", "compute_cdi.py", "ndarray"),
    ):
        path = SIBLING_PROTOTYPES / sib / fname
        try:
            mod = _import_sibling(f"sibling_{sib}", path)
            fn = mod.hazen_percentiles
            corpus = corpora["ties_mixed"]
            if adapter == "dict":
                keyed = {f"k{i:02d}": v for i, v in enumerate(corpus)}
                got_map = fn(keyed, True)
                got = [got_map[f"k{i:02d}"] for i in range(len(corpus))]
            else:
                import numpy as np
                got = [float(x) for x in fn(np.array(corpus, dtype=float))]
            match = all(abs(a - b) < 1e-9 for a, b in zip(got, expected["ties_mixed"]))
            detail[sib] = {"values": [round(v, 4) for v in got], "matches_expected": match}
            ok = ok and match
        except Exception as exc:  # noqa: BLE001 — parity must not crash the harness
            detail[sib] = {"skipped": f"{type(exc).__name__}: {exc}"}
            ok = False
    return {"pass": ok, "corpus": corpora["ties_mixed"], "expected": expected["ties_mixed"], "detail": detail,
            "rule": "one family transform: ADI, SDI, CDI implementations must agree on a tie-containing corpus"}


def check_coverage(output: Dict) -> Dict[str, object]:
    rows = output["data"]
    qs = [r["quarter"] for r in rows]
    contiguous = all(q_index(b) - q_index(a) == 1 for a, b in zip(qs, qs[1:]))
    mp_bounds = {}
    for d in compute_adi.DOMAIN_IDS:
        counts = [r["domains"][d]["members_present"] for r in rows]
        mp_bounds[d] = {"min": min(counts), "max": max(counts), "registry_members": len(compute_adi.DOMAIN_MEMBERS[d])}
    # Trailing quarters with data but no publication, and the domains that block them
    loaded = compute_adi.load_inputs(REPO_ROOT)
    result = compute_adi.build(loaded)
    last_pub = q_index(qs[-1])
    all_q = sorted({q for m in result.member_pct.values() for q in m})

    # Data-derived expected start: the engine publishes a quarter only when all
    # five domains have >=1 member present (compute_adi step 7). The earliest
    # such quarter IS the only correct first published quarter. Deriving it
    # from the inputs — instead of pinning a hardcoded "2005-Q1" — means this
    # gate follows the source history automatically (it was the 2005-Q1 pin
    # that masked the truncated-history bug: the floor and the assertion
    # agreed on the wrong answer). The gate still fails on the real faults:
    # a gap (contiguity), publishing a quarter before full domain coverage,
    # or skipping the first fully-covered quarter.
    def _all_domains_present(q: str) -> bool:
        return all(
            any(q in result.member_pct[m] for m in compute_adi.DOMAIN_MEMBERS[d])
            for d in compute_adi.DOMAIN_IDS
        )
    fully_covered = [q for q in all_q if _all_domains_present(q)]
    expected_start = fully_covered[0] if fully_covered else None

    trailing = []
    for q in all_q:
        if q_index(q) <= last_pub:
            continue
        present = {d for d in compute_adi.DOMAIN_IDS
                   if any(q in result.member_pct[m] for m in compute_adi.DOMAIN_MEMBERS[d])}
        trailing.append({"quarter": q, "domains_present": sorted(present),
                         "domains_missing": sorted(set(compute_adi.DOMAIN_IDS) - present)})
    return {
        "pass": contiguous and qs[0] == expected_start and expected_start is not None,
        "published": f"{qs[0]} to {qs[-1]} ({len(qs)} quarters, contiguous={contiguous})",
        "expected_start_from_inputs": expected_start,
        "start_matches_first_fully_covered_quarter": qs[0] == expected_start,
        "members_present_bounds": mp_bounds,
        "trailing_unpublished": trailing,
    }


# ---------------------------------------------------------------------------


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    checks: Dict[str, Dict[str, object]] = {}
    checks["cross_process_reproducibility"] = check_cross_process_reproducibility()
    output = json.loads(OUTPUT_JSON.read_text())
    checks["engine_gates"] = check_engine_gates(output)
    checks["seam_independent"] = check_seam_independent(output)
    checks["gfc_independent"] = check_gfc_independent(output)
    checks["orientation"] = check_orientation()
    checks["weights"] = check_weights(output)
    checks["hazen_parity"] = check_hazen_parity()
    checks["coverage"] = check_coverage(output)

    REPORT_PATH.parent.mkdir(parents=True, exist_ok=True)
    REPORT_PATH.write_text(json.dumps(checks, indent=2, ensure_ascii=False) + "\n")

    all_pass = True
    for name, c in checks.items():
        verdict = "PASS" if c["pass"] else "FAIL"
        logger.info("%-32s %s", name, verdict)
        if not c["pass"]:
            all_pass = False
    logger.info("report: %s", REPORT_PATH)
    return 0 if all_pass else 1


if __name__ == "__main__":
    raise SystemExit(main())