#!/usr/bin/env python3
"""Independent validator for the family-v1 American Worker Index.

Mirrors validate_adi.py: re-runs the engine cross-process to prove
byte-reproducibility, re-reads the committed worker_index.json and asserts
every engine gate passed, and independently recomputes the load-bearing
structural facts (the pre/post-1980 regime break, the peak quarter, the
publication contiguity, the trailing full-member invariant) from the published
rows rather than trusting the engine's own numbers.

Run:  PYTHONPATH=. python3 scripts/indexes/validate_worker_index.py
Writes data/audit/index_family/worker_index_validation.json.
Exit 0 only when every check passes.
"""
from __future__ import annotations

import hashlib
import json
import logging
import subprocess
import sys
from pathlib import Path
from typing import Dict, List

logger = logging.getLogger("validate_worker_index")

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 import compute_worker_index as W  # noqa: E402

ENGINE = REPO_ROOT / "scripts" / "indexes" / "compute_worker_index.py"
JSON_OUT = REPO_ROOT / "data" / "indexes" / "worker_index.json"
CSV_OUT = REPO_ROOT / "data" / "indexes" / "worker_index.csv"
PUBLIC_JSON = REPO_ROOT / "site" / "public" / "data" / "indexes" / "worker_index.json"
PUBLIC_CSV = REPO_ROOT / "site" / "public" / "data" / "indexes" / "worker_index.csv"
REPORT = REPO_ROOT / "data" / "audit" / "index_family" / "worker_index_validation.json"


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


def check_cross_process_reproducibility() -> Dict[str, object]:
    """Run the engine as two separate subprocesses; the four written artifacts
    must be byte-identical across runs."""
    hashes: List[Dict[str, str]] = []
    for _ in range(2):
        subprocess.run([sys.executable, str(ENGINE)], cwd=str(REPO_ROOT), check=True,
                       capture_output=True, env={"PYTHONPATH": str(REPO_ROOT), "PATH": ""})
        hashes.append({
            "json": _sha(JSON_OUT), "csv": _sha(CSV_OUT),
            "public_json": _sha(PUBLIC_JSON), "public_csv": _sha(PUBLIC_CSV),
        })
    ok = hashes[0] == hashes[1]
    return {"pass": ok, "run1": hashes[0], "run2": hashes[1],
            "rule": "two separate engine subprocesses write byte-identical artifacts"}


def check_public_mirror_parity() -> Dict[str, object]:
    ok = (JSON_OUT.read_bytes() == PUBLIC_JSON.read_bytes()
          and CSV_OUT.read_bytes() == PUBLIC_CSV.read_bytes())
    return {"pass": ok,
            "rule": "the public download mirror is byte-identical to the canonical output"}


def check_engine_gates(doc: dict) -> Dict[str, object]:
    gates = doc.get("validation", {})
    failed = [name for name, g in gates.items() if not g.get("pass")]
    return {"pass": not failed, "failed": failed,
            "rule": "every validation[*].pass recorded in worker_index.json is true"}


def check_regime_break(doc: dict) -> Dict[str, object]:
    """Independently recompute the pre/post-1980 mean split from the published
    rows and require it to be a real, positive break (the load-bearing finding)."""
    rows = doc["data"]
    pre = [r["composite"] for r in rows if r["quarter"] < "1980"]
    post = [r["composite"] for r in rows if r["quarter"] >= "1980"]
    if not pre or not post:
        return {"pass": False, "rule": "need pre- and post-1980 quarters"}
    gap = sum(post) / len(post) - sum(pre) / len(pre)
    return {"pass": gap > 15.0, "pre_1980_mean": round(sum(pre) / len(pre), 2),
            "post_1980_mean": round(sum(post) / len(post), 2), "gap": round(gap, 2),
            "rule": "post-1980 mean exceeds pre-1980 mean by > 15 points (the worker-decline regime break)"}


def check_peak_and_contiguity(doc: dict) -> Dict[str, object]:
    """Recompute the peak quarter and publication contiguity from the rows and
    match them to the engine's recorded summary."""
    rows = doc["data"]
    quarters = [r["quarter"] for r in rows]
    idx = [W.quarter_index(q) for q in quarters]
    contiguous = all(b - a == 1 for a, b in zip(idx, idx[1:]))
    peak_row = max(rows, key=lambda r: r["composite"])
    recorded_peak = doc["summary"]["peak"]["quarter"]
    tail_full = W._has_full_member_set(quarters[-1], _member_pct_from_doc(doc))
    ok = contiguous and peak_row["quarter"] == recorded_peak and tail_full
    return {"pass": ok, "recomputed_peak": peak_row["quarter"], "recorded_peak": recorded_peak,
            "published_contiguous": contiguous, "tail_has_full_member_set": tail_full,
            "rule": "published range contiguous AND the recomputed peak equals the recorded "
                    "summary peak AND the last published quarter carries the full member set"}


def _member_pct_from_doc(doc: dict) -> Dict[str, Dict[str, float]]:
    """Reconstruct the member->quarter->pct map from the published rows so the
    trailing-full-member invariant can be re-checked without re-running build."""
    out: Dict[str, Dict[str, float]] = {s.input_id: {} for s in W.WORKER_REGISTRY}
    for r in doc["data"]:
        for iid, pct in r.get("members", {}).items():
            out.setdefault(iid, {})[r["quarter"]] = pct
    return out


def check_hazen_parity() -> Dict[str, object]:
    """The worker engine's Hazen transform must agree with the shared family
    implementation on a fixed tie-containing corpus (one yardstick for all)."""
    corpus = [3.2, 1.1, 3.2, 0.5, 9.9, 3.2, 1.1, 7.0]
    expected = [56.25, 25.0, 56.25, 6.25, 93.75, 56.25, 25.0, 81.25]
    got = W.hazen_percentiles_series(corpus)
    ok = all(abs(a - b) < 1e-9 for a, b in zip(got, expected))
    return {"pass": ok, "expected": expected, "got": got,
            "rule": "the shared Hazen transform matches the fixed family parity corpus to 1e-9"}


def main() -> int:
    doc = json.loads(JSON_OUT.read_text())
    checks = {
        "cross_process_reproducibility": check_cross_process_reproducibility(),
        "public_mirror_parity": check_public_mirror_parity(),
        "engine_gates": check_engine_gates(doc),
        "regime_break": check_regime_break(doc),
        "peak_and_contiguity": check_peak_and_contiguity(doc),
        "hazen_parity": check_hazen_parity(),
    }
    # re-read after the subprocess reruns (they rewrote the file identically)
    doc = json.loads(JSON_OUT.read_text())
    checks["engine_gates"] = check_engine_gates(doc)
    checks["regime_break"] = check_regime_break(doc)
    checks["peak_and_contiguity"] = check_peak_and_contiguity(doc)

    all_pass = all(c["pass"] for c in checks.values())
    REPORT.parent.mkdir(parents=True, exist_ok=True)
    REPORT.write_text(json.dumps({"all_pass": all_pass, "checks": checks}, indent=2) + "\n")

    for name, c in checks.items():
        logger.info("%-4s  %s", "PASS" if c["pass"] else "FAIL", name)
    logger.info("\n%s", "ALL CHECKS PASS" if all_pass else "VALIDATION FAILED")
    return 0 if all_pass else 1


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    raise SystemExit(main())