SniperGold_ML/engine/verify/runner.py

"""Independent verifier runner (Layer 12, spec 18).

Provides the verification entry points used by the CLI and by the
qualification suite: golden corpus (producer == verifier), dataset mode
(content ids recomputed independently), run-complete mode, and invariant runs.
"""

import json
import os

from .. import versions
from ..canonical import tick_chunk_content_id, bar_file_content_id
from ..manifest import load_manifest, verify_manifest, dataset_hash_from_files
from ..storage import (
    ticks_chunk_path, bar_part_path, read_ticks_chunk, read_bar_part,
    cert_path, checkpoint_path, chunkmap_path,
)
from ..util import atomic_write_json, sha256_file

VERDICT_ACCEPTED = "VERIFIER_ACCEPTED"
VERDICT_REJECTED = "VERIFIER_REJECTED"


def _report_path(out_root, run_id):
    return os.path.join(out_root, "verification", "report_%s.json" % run_id)


def write_report(out_root, run_id, checks, verdict):
    payload = {"verdict": verdict, "checks": checks}
    atomic_write_json(_report_path(out_root, run_id), payload)
    return payload


def verify_ticks_content_ids_against_evidence(out_root, chunkmap, evidence):
    """Independent recomputation of every ticks-chunk content id from disk
    (vinvariants.independent_ticks_content_id) and comparison with the
    evidence records captured at production time."""
    from .vinvariants import independent_ticks_content_id
    diffs = []
    for c in chunkmap["chunks"]:
        rel = "ticks/chunk_%06d.parquet" % c["index"]
        if rel not in evidence.get("files", {}):
            diffs.append("missing in evidence: " + rel)
            continue
        rows = read_ticks_chunk(ticks_chunk_path(out_root, c["index"]))
        actual = independent_ticks_content_id(
            [(r[0], r[1], r[2], r[4]) for r in rows])
        if actual != evidence["files"][rel].get("content_id"):
            diffs.append("content_id mismatch: " + rel)
    return (not diffs), diffs


def verify_bars_content_ids_against_evidence(out_root, chunkmap, cfg, evidence):
    from .vinvariants import independent_bars_content_id
    diffs = []
    for wl in chunkmap["workloads"]:
        for tf in cfg["timeframes"]:
            rel = "bars/%s/wl_%03d.parquet" % (tf, wl["index"])
            p = bar_part_path(out_root, tf, wl["index"])
            if not os.path.exists(p):
                continue
            rows = read_bar_part(p)
            actual = independent_bars_content_id(rows)
            if actual != evidence["files"][rel].get("content_id"):
                diffs.append("bar content_id mismatch: " + rel)
    return (not diffs), diffs


def verify_preservation_against_evidence(out_root, chunkmap, evidence):
    """Independent recomputation of every source-preservation sidecar content
    id + canonical row-count / src_line order cross-check (spec 8.6, 18.3).
    The sidecar exists only for G_TICKSTORY_MT5 runs; absence is a pass."""
    from .vinvariants import independent_preservation_content_id
    from ..storage import preservation_chunk_path, ticks_chunk_path, \
        read_ticks_chunk
    diffs = []
    for c in chunkmap["chunks"]:
        rel = "source_preservation/chunk_%06d.jsonl" % c["index"]
        pp = preservation_chunk_path(out_root, c["index"])
        if not os.path.exists(pp):
            continue
        if rel not in evidence.get("files", {}):
            diffs.append("missing in evidence: " + rel)
            continue
        with open(pp, "r", encoding="utf-8") as fh:
            lines = fh.read().splitlines()
        pairs = []
        for line in lines:
            line = line.strip()
            if not line:
                continue
            s, l = line.split("|", 1)
            pairs.append((int(s), int(l)))
        actual = independent_preservation_content_id(pairs)
        if actual != evidence["files"][rel].get("content_id"):
            diffs.append("content_id mismatch: " + rel)
        rows = read_ticks_chunk(ticks_chunk_path(out_root, c["index"]))
        if len(pairs) != len(rows):
            diffs.append("row-count mismatch: " + rel)
        for pair, row in zip(pairs, rows):
            if pair[0] != row[5]:
                diffs.append("src_line order mismatch: " + rel)
                break
    return (not diffs), diffs


def verify_dataset(out_root, chunkmap, cfg, run_id):
    """Verify --mode dataset: recompute every content id independently and the
    canonical dataset hash from artifacts."""
    checks = []
    evidence_path = os.path.join(out_root, "evidence", "evidence.json")
    with open(evidence_path, "r", encoding="utf-8") as fh:
        evidence = json.load(fh)
    ok1, d1 = verify_ticks_content_ids_against_evidence(out_root, chunkmap, evidence)
    ok2, d2 = verify_bars_content_ids_against_evidence(out_root, chunkmap, cfg, evidence)
    ok3, d3 = verify_preservation_against_evidence(out_root, chunkmap, evidence)
    checks.append({"check": "ticks_content_ids_independent", "result": "PASS" if ok1 else "FAIL",
                   "detail": d1})
    checks.append({"check": "bars_content_ids_independent", "result": "PASS" if ok2 else "FAIL",
                   "detail": d2})
    checks.append({"check": "preservation_content_ids_independent",
                   "result": "PASS" if ok3 else "FAIL", "detail": d3})
    verdict = VERDICT_ACCEPTED if (ok1 and ok2 and ok3) else VERDICT_REJECTED
    return write_report(out_root, run_id, checks, verdict)


def verify_run_complete(out_root, run_id):
    """Verify --mode run-complete: re-check RUN_COMPLETE.json + evidence."""
    from ..evidence import verify_evidence, load_evidence
    from ..run_complete import verify_run_complete as vrc
    checks = []
    ev_ok, ev_diffs = verify_evidence(out_root)
    evidence = load_evidence(out_root)
    rc_ok, rc_diffs = vrc(out_root, evidence)
    checks.append({"check": "evidence_manifest", "result": "PASS" if ev_ok else "FAIL",
                   "detail": ev_diffs})
    checks.append({"check": "run_complete", "result": "PASS" if rc_ok else "FAIL",
                   "detail": rc_diffs})
    verdict = VERDICT_ACCEPTED if (ev_ok and rc_ok) else VERDICT_REJECTED
    return write_report(out_root, run_id, checks, verdict)


def verify_golden(cases_dir, expected_dir, use_spawn=False):
    """Verify --mode golden: run the golden corpus through both the producer
    and the independent verifier and require byte-equal expected outputs.
    Delegates to the golden runner in tests/golden/run_golden.py."""
    from tests.golden.run_golden import run_golden
    return run_golden(cases_dir=cases_dir, expected_dir=expected_dir)


def run_invariants(records, bar_rows_by_tf):
    """Run the independent invariant checks (spec 18.3) on a merged stream."""
    from .vinvariants import (
        check_monotonic_stream, check_ohlc_validity, check_bar_uniqueness,
    )
    checks = []
    monotonic, md = check_monotonic_stream(records)
    checks.append({"check": "merged_stream_monotonic", "result": "PASS" if monotonic else "FAIL",
                   "detail": md})
    for tf, rows in bar_rows_by_tf.items():
        ohlc, od = check_ohlc_validity(rows)
        uniq, ud = check_bar_uniqueness(rows)
        checks.append({"check": "ohlc_validity_%s" % tf,
                       "result": "PASS" if ohlc else "FAIL", "detail": od})
        checks.append({"check": "bar_uniqueness_%s" % tf,
                       "result": "PASS" if uniq else "FAIL", "detail": ud})
    return checks