SniperGold_ML/tests/legacy_diff/run_legacy_diff.py

"""Differential legacy comparison (spec 24, G-16) -- evidence-level only.

The legacy P3-S25 evidence (per-chunk JSON + real checkpoint) is read
read-only; every compared item is classified into the five-class taxonomy and
no UNRESOLVED semantic discrepancy may remain. The real 34.5 GB source is
NOT opened in this session (implementation + qualification only); the
byte-range/tick/hash differential is recorded as the G-13 pilot scope.
"""

import json
import os
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__)))))

from engine.verify.vcompare import (  # noqa: E402
    legacy_evidence_items, build_legacy_diff, write_legacy_diff, Taxonomy,
)
from engine.util import sha256_file  # noqa: E402
from tests.golden.run_golden import mini_setup, mini_run  # noqa: E402
from tests.golden.common import case_bytes  # noqa: E402

REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
LEGACY_INGEST = os.path.join(REPO, "ml", "p3", "p3_s251_external_ingest")
LEGACY_CHUNK_GLOB = "real_run_chunk_*.json"
LEGACY_CHECKPOINT = os.path.join(LEGACY_INGEST, "output", "s251_real_checkpoint.json")


def run():
    # read-only snapshot of the legacy evidence dir hashes (byte-unchanged check)
    before = {}
    for f in sorted(os.listdir(LEGACY_INGEST)):
        if f.startswith("real_run_chunk") and f.endswith(".json"):
            before[f] = sha256_file(os.path.join(LEGACY_INGEST, f))
    before_cp = sha256_file(LEGACY_CHECKPOINT)

    chunk_files = sorted(os.path.join(LEGACY_INGEST, f) for f in before)
    legacy = legacy_evidence_items(chunk_files, LEGACY_CHECKPOINT)

    # new engine metadata from a synthetic mini run (permitted fixture)
    with tempfile.TemporaryDirectory() as td:
        cfg, cert, cm, rid = mini_setup(case_bytes("G01"), td,
                                        chunk_bytes=25_165_824,
                                        workload_bytes=5_368_709_120)
        mini_run(cfg, cert, cm, td, rid)
        from engine.storage import checkpoint_path
        from engine.checkpoint import load_checkpoint
        ck = load_checkpoint(checkpoint_path(td))
        report = build_legacy_diff(legacy, ck, ck["versions"],
                                   cfg["workers_requested"], 1)

    # verify no UNRESOLVED items and the DEFERRED scope marker
    unresolved = [i for i in report["items"]
                  if i["classification"] == Taxonomy.UNRESOLVED]
    deferred = report.get("real_source_range_differential", {}).get("status")
    all_classified = len(report["items"]) > 0 and len(unresolved) == 0

    # verify legacy evidence untouched
    after = {f: sha256_file(os.path.join(LEGACY_INGEST, f)) for f in before}
    after_cp = sha256_file(LEGACY_CHECKPOINT)
    untouched = before == after and before_cp == after_cp

    return {
        "ok": all_classified and deferred == "DEFERRED" and untouched,
        "items_classified": len(report["items"]),
        "unresolved": len(unresolved),
        "deferred_status": deferred,
        "legacy_untouched": untouched,
        "report": report,
    }


if __name__ == "__main__":
    res = run()
    print("items_classified", res["items_classified"])
    print("unresolved", res["unresolved"])
    print("deferred_status", res["deferred_status"])
    print("legacy_untouched", res["legacy_untouched"])
    print("OK" if res["ok"] else "FAIL")
    sys.exit(0 if res["ok"] else 1)