SniperGold_ML/tests/qualification_main.py

"""P3-DATA-ENGINE-002 qualification driver.

Runs every mandatory gate: unit tests, golden corpus G01-G17, property tests,
adversarial tests, mutation tests (per-mutation), independent verifier modes
(dataset + run-complete on a synthetic run), differential legacy comparison,
research dataset builder determinism, and headless CLI qualification.

Produces qualification/QUALIFICATION_REPORT.json and .md.
Verdict is exactly one of: QUALIFIED | NOT QUALIFIED.
"""

import datetime
import json
import os
import sys
import tempfile

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

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
QUAL_DIR = os.path.join(REPO, "qualification")

# Non-routine interventions ledger (mandatory accounting). Every entry is an
# actual event from this implementation session.
INTERVENTIONS = [
    {
        "trigger": "Producer/verifier epoch mismatch caught by dual-path tests",
        "observed_evidence": "parse_timestamp_dotted returned 2021 epoch for a 2023 timestamp; independent table-based verifier disagreed",
        "finding": "days_from_civil month-offset constant used the wrong value (1/-3 instead of 9/-3)",
        "decision_basis": "spec 8.1 requires deterministic integer epoch math; both paths must agree exactly",
        "action": "Corrected the Hinnant civil-days formula; re-verified producer == verifier on all fixtures",
        "result": "Golden G01..G12 and property content-id stability PASS",
        "state_impact": "None outside engine/util.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Bar content-id raises on 17-field CBS rows",
        "observed_evidence": "serialize_bars expected 14 fields, storage rows carry 17",
        "finding": "spec 12.4 canonical bar serialization defines exactly 14 fields; start_ms/end_ms/spread_avg_u are not serialized",
        "decision_basis": "spec 12.4 field list is authoritative",
        "action": "Producer and verifier serializers now select the 14 canonical fields from full CBS rows",
        "result": "All bar content-id checks PASS",
        "state_impact": "engine/canonical.py + engine/verify/vaggregate.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Resume re-aggregates an already processed workload",
        "observed_evidence": "Pause->resume on a workload with no finalized bars double-counted carry ticks; carry-continuity FAILED",
        "finding": "Aggregation completeness was tracked only by bar-file presence, but zero-bar workloads wrote no file",
        "decision_basis": "append-only outputs (spec 4.2/11.5) and resume equivalence require exactly-once aggregation per workload",
        "action": "Every processed workload now writes a per-TF part file (empty when it contains no finalized bars) as an aggregation-completeness marker; resume skips workloads whose parts exist",
        "result": "Repeated-resume and resume-equivalence tests PASS; no bar rows were ever fabricated",
        "state_impact": "engine/dispatcher.py; documented in implementation map",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Journal chain broke at seq 1 during read-back",
        "observed_evidence": "append_commit stored hash of line including trailing \\n; read_journal recomputed without it",
        "finding": "Inconsistent line-hash domain",
        "decision_basis": "spec 13.1 hash-chained journal; deterministic chain definition required",
        "action": "Standardized the line hash over the line text without the trailing newline",
        "result": "Journal round-trip and resume checks PASS",
        "state_impact": "engine/journal.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "pyarrow replace_schema_metadata TypeError (expected bytes, int found)",
        "observed_evidence": "int metadata values rejected by pyarrow 25",
        "finding": "Parquet schema key-value metadata must be strings",
        "decision_basis": "spec 12.3 schema metadata is recorded as text",
        "action": "Stringify all schema metadata values",
        "result": "All parquet write/read paths PASS",
        "state_impact": "engine/storage.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Spec gap: carry field list (§11.4) omits first_src_line needed by CBS_V1",
        "observed_evidence": "A partial bar carried across a resume would lose its first_src_line audit anchor",
        "finding": "first_src_line must survive resume to keep CBS_V1 semantics and resume-equivalent content ids",
        "decision_basis": "spec 11.3/13.5 require the audit anchor and resume equivalence; checkpoint-internal field addition is the minimal deterministic resolution and changes no external contract",
        "action": "Checkpointed carry objects additionally store first_src_line; documented in the implementation map",
        "result": "Resume-equivalence content ids stable",
        "state_impact": "engine/aggregate.py (documented)",
        "authorization_status": "authorized (implementation detail; no spec amendment)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "CLI pause/stop lacked --out-root; verify --mode dataset raced final checkpoint writes",
        "observed_evidence": "pause returned usage error 2; run-complete re-verification found stale state hashes",
        "finding": "Parser gaps and evidence build before final checkpoint",
        "decision_basis": "spec 14 exit codes and 22 completion protocol ordering",
        "action": "Added --out-root to pause/stop; reordered finalize to verify -> COMPLETED -> evidence -> RUN_COMPLETE",
        "result": "CLI E2E 14/14 PASS with correct exit codes",
        "state_impact": "engine/cli.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    # ---- P3-DATA-ENGINE-004 (source-grammar correction + re-qualification) --
    {
        "trigger": "P3-DE-003 pilot audit: the real six-column Tickstory grammar differs from the dotted corpus grammar",
        "observed_evidence": "First real row '20030505,00:01:03,340.345,340.757,340.345,0'; real source has no header, CRLF, explicit volume column (zeros valid) and a 'last' price column",
        "finding": "Spec 8.1 primary grammar was the dotted corpus form; the authoritative production grammar is G_TICKSTORY_MT5 (YYYYMMDD,HH:MM:SS,bid,ask,last,volume)",
        "decision_basis": "Owner-authorized controlled amendment (P3-DE-004); Appendix C supersedes the dotted primary for the production source",
        "action": "Amended spec Appendix C (C.1 grammar, C.2 timestamp, C.3 volume/last/versions, C.4 implementation-visible changes, C.5 exclusions); specification SHA updated to effbaf2624cd46137c22a246fdabff4067052807b6b6224be464366bec4745d9",
        "result": "Spec 1.1.0 carries the corrected grammar; G-9/G-10/G-14 unchanged; engine version identity bumped (1.1.0 / PARSER_V1.1)",
        "state_impact": "docs/P3_DATA_ENGINE_V1_SPEC.md only",
        "authorization_status": "authorized (approved amendment package)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Grammar presets, compact timestamp and the 'last' column required implementation changes",
        "observed_evidence": "Dotted parser cannot recognize compact dates; 'last' has no CTS_V1 mapping (spec 8.6)",
        "finding": "Need a named G_TICKSTORY_MT5 preset (certificate-pinned), compact civil-math timestamp, mandatory volume semantics and a source-preservation layer",
        "decision_basis": "Spec 8.1/8.2/8.6 + Appendix C.3; CTS_V1/CBS_V1 schema identity must stay frozen (G-4/G-5)",
        "action": "versions registry + version bumps; config source_grammar identity field + fail-closed validation; parse_compact_datetime; _parse_chunk_tickstory; preservation serialization + per-chunk sidecar writes; certificate grammar_id + compact sniffing; init certificate/config mismatch BLOCK; independent verifier v_parse_chunk_tickstory + preservation id checks; evidence + verify_dataset preservation content-id verification",
        "result": "SG01-SG15 + compact E2E PASS; producer == independent verifier on all fixtures; anchor row 20030505,00:01:03 -> ts 1,052,092,863,000 ms (legacy-verified)",
        "state_impact": "engine/{parse,canonical,certify,config,dispatcher,evidence,storage,versions,worker}.py; engine/verify/{runner,vinvariants,vparse}.py; tests/source_grammar_tests.py (new)",
        "authorization_status": "authorized (implementation of the approved amendment)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "P3-DE-003 independent pilot audit: CLI status/resume on a run-less directory misbehaves",
        "observed_evidence": "Expected exit code 4 (BLOCKED); actual exit code 1 + TypeError traceback",
        "finding": "_load_run_context returned the error integer in the last tuple slot; callers dereferenced a None config (NEW_ENGINE_DEFECT, non-data-affecting)",
        "decision_basis": "Spec 14.3 exit-code contract (4 = blocked); P3-DE-004 Appendix C.4",
        "action": "Return the _err code as the first tuple element; callers return it; checkpoint-load exceptions become exit 4; regression tests cli_status_runless_dir / cli_resume_runless_dir added",
        "result": "CLI E2E regression PASS (exit 4 without traceback)",
        "state_impact": "engine/cli.py; tests/cli_tests.py",
        "authorization_status": "authorized (implementation)",
        "necessary_or_avoidable": "necessary",
    },
    {
        "trigger": "Resume-session inspection found a broken partial edit in the qualification driver",
        "observed_evidence": "The previous session's import edit REPLACED 'from tests import property_tests, adversarial_tests' with 'from tests import source_grammar_tests'; property/adversarial gates would NameError and no source_grammar gate existed",
        "finding": "Partial edit; the suite must keep property/adversarial AND add source grammar",
        "decision_basis": "Do not rewrite the driver; restore both imports, add the source_grammar gate, refresh report fields",
        "action": "Restored both import lines; added gates['source_grammar'] (suite_cases + e2e); updated session/SHA/governance/versions/safety/limitations/interventions and the Markdown header",
        "result": "Full qualification discovers and runs every suite including SG01-SG15",
        "state_impact": "tests/qualification_main.py",
        "authorization_status": "authorized (implementation; resume within P3-DE-004 scope)",
        "necessary_or_avoidable": "necessary",
    },
]


def _utc_now():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def run_qualification(quick=False):
    from tests import unit_tests
    from tests.golden.run_golden import run_golden
    from tests import property_tests, adversarial_tests
    from tests import source_grammar_tests
    from engine import versions
    from tests.mutations.run_mutations import run as run_mutations
    from tests.legacy_diff.run_legacy_diff import run as run_legacy_diff
    from tests.dataset_builder_tests import run as run_dataset_builder
    from tests.cli_tests import run as run_cli

    gates = {}

    # 1 unit tests ---------------------------------------------------------
    unit_ok = unit_tests.run()
    gates["unit"] = {"pass": unit_ok, "note": "29 unit cases (12 areas)"}

    # 2 golden corpus -------------------------------------------------------
    golden = run_golden()
    gates["golden"] = {
        "pass": golden["verdict"] == "VERIFIER_ACCEPTED",
        "cases": golden["cases"],
    }

    # 2b source-grammar golden cases (SG01-SG15) + compact E2E --------------
    # P3-DE-004: the revised engine must independently pass the six-column
    # Tickstory fixtures; the producer and the independent verifier must
    # agree on every canonical row, malformed class and preservation id.
    sg = source_grammar_tests.run()
    gates["source_grammar"] = {
        "pass": sg["ok"],
        "suite_cases": sg["suite_cases"],
        "e2e": sg["e2e"],
    }

    # 3 property/invariant tests -------------------------------------------
    prop = property_tests.run()
    ok = all(r[1] for r in prop)
    gates["property"] = {"pass": ok,
                          "items": [{"name": n, "pass": o, "detail": "%s" % d}
                                    for n, o, d in prop]}

    # 4 adversarial tests ----------------------------------------------------
    adv = adversarial_tests.run()
    ok = all(r[1] for r in adv)
    gates["adversarial"] = {"pass": ok,
                             "items": [{"name": n, "pass": o, "detail": "%s" % d}
                                       for n, o, d in adv]}

    # 5 mutation tests ------------------------------------------------------
    mut_ledger = run_mutations()
    ok = all(m["detected"] for m in mut_ledger)
    gates["mutation"] = {"pass": ok, "detected": sum(1 for m in mut_ledger
                                                     if m["detected"]),
                          "total": len(mut_ledger), "ledger": mut_ledger}

    # 6 independent verifier modes (dataset + run-complete) -----------------
    with tempfile.TemporaryDirectory(dir=QUAL_DIR) as td:
        from tests.golden.run_golden import mini_setup, mini_run, _file_ids
        from tests.golden.common import case_bytes
        from engine.storage import checkpoint_path
        from engine.checkpoint import load_checkpoint
        from engine.evidence import build_evidence, verify_evidence
        from engine.verify.runner import verify_dataset, \
            verify_run_complete as vrc_mode
        from engine.dispatcher import finalize_completed
        from engine.run_complete import build_run_complete
        from engine.cli import _finalize  # noqa: F401 (ordering helper)
        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)
        ck = load_checkpoint(checkpoint_path(td))
        # independent recomputation path (producer independent verifier)
        build_evidence(td, cm, cfg)
        rep_d = verify_dataset(td, cm, cfg, rid)
        # finalize + run-complete
        finalize_completed(cfg, td, rid)
        evidence = build_evidence(td, cm, cfg)
        build_run_complete(td, ck, evidence,
                           os.path.join(td, "verification", "report_%s.json" % rid),
                           rep_d["verdict"],
                           {rel: rec["content_id"] for rel, rec
                            in evidence["files"].items() if rec.get("content_id")},
                           {"ticks_files": 1, "bar_files_per_TF": {t: 1
                                                                  for t in cfg["timeframes"]}})
        rep_r = vrc_mode(td, rid)
        ev_ok, ev_diffs = verify_evidence(td)
        ver_ok = (rep_d["verdict"] == "VERIFIER_ACCEPTED"
                  and rep_r["verdict"] == "VERIFIER_ACCEPTED" and ev_ok)
        gates["independent_verifier"] = {
            "pass": ver_ok,
            "dataset_verdict": rep_d["verdict"],
            "run_complete_verdict": rep_r["verdict"],
            "evidence_recheck": ev_ok,
            "evidence_diffs": ev_diffs,
        }

    # 7 differential legacy comparison ------------------------------------
    legacy = run_legacy_diff()
    gates["legacy_diff"] = {
        "pass": legacy["ok"],
        "items_classified": legacy["items_classified"],
        "unresolved": legacy["unresolved"],
        "deferred_status": legacy["deferred_status"],
        "legacy_untouched": legacy["legacy_untouched"],
    }

    # 8 research dataset builder determinism --------------------------------
    ds = run_dataset_builder()
    gates["dataset_builder"] = {"pass": ds["ok"],
                                 "hash_stable": ds["dataset_hash_stable"],
                                 "n_bars": ds["n_bars"],
                                 "excluded_no_future": ds["excluded_no_future"]}

    # 9 CLI qualification -----------------------------------------------------
    cli = run_cli()
    gates["cli"] = {"pass": cli["ok"], "results": cli["results"],
                     "run_complete_created": cli["run_complete_created"]}

    all_pass = all(gates[k]["pass"] for k in gates)
    verdict = "QUALIFIED" if all_pass else "NOT QUALIFIED"

    report = {
        "session": "P3-DATA-ENGINE-004",
        "generated_utc": _utc_now(),
        "specification_doc": "docs/P3_DATA_ENGINE_V1_SPEC.md",
        "specification_sha256": "effbaf2624cd46137c22a246fdabff4067052807b6b6224be464366bec4745d9",
        "governance_baseline": "P3-DATA-ENGINE-001 GOV-CLOSE-001 (DESIGN FROZEN) + P3-DE-004 CONTROLLED AMENDMENT (spec 1.1.0, Appendix C); G-14 NOT AUTHORIZED",
        "versions": {
            "spec_version": "1.1.0 (amended by P3-DATA-ENGINE-004; Appendix C)",
            "engine_version": versions.ENGINE_VERSION,
            "parser_version": versions.PARSER_VERSION,
            "algorithm_version": versions.ALGORITHM_VERSION,
            "schema": "CTS_V1 / CBS_V1 (unchanged)",
            "dataset_version": versions.DATASET_VERSION,
            "grammar_registry": list(versions.SUPPORTED_GRAMMARS),
        },
        "gates": gates,
        "verdict": verdict,
        "interventions": INTERVENTIONS,
        "safety": {
            "34gb_source_processed": "NO",
            "real_data_pilot_executed": "NO",
            "real_source_ingested": "NO",
            "workload_46_executed": "NO",
            "chunk_760_plus_touched": "NO",
            "legacy_checkpoint_modified": "NO",
            "legacy_pipeline_modified": "NO",
            "specification_amendment_applied": "YES (approved P3-DE-004 controlled amendment; Appendix C; SHA verified)",
            "governance_baseline_altered": "NO (unchanged except the approved specification amendment)",
            "g14_authorization_changed": "NO (G-14 remains BLOCKED / NOT AUTHORIZED)",
            "forecasting_performed": "NO",
            "research_dataset_generated": "NO",
            "production_checkpoint_created": "NO",
        },
        "safety_notes": {
            "research_dataset_generated": "NO: no real/research dataset was produced. The dataset-builder and build-dataset qualification ran on synthetic fixtures only (permitted by session scope).",
            "source_grammar_correction": "The P3-DE-004 source-grammar correction (G_TICKSTORY_MT5 six-column preset) is an approved specification amendment, not a legacy defect. Canonical schema/algorithm semantics were not re-opened (G-4/G-5 unchanged).",
        },
        "limitations": [
            "NOT VERIFIED: byte-range/tick/hash differential vs the real 34.5 GB source (G-13 pilot scope; session forbids opening the source).",
            "NOT VERIFIED: full 34 GB headless pass (G-14 NOT AUTHORIZED).",
            "DEFERRED: G-17 HMAC hardening.",
            "PARTIAL: CLI 'verify --mode range/legacy' live-range modes exercised only on synthetic fixtures.",
            "PARTIAL: the source-preservation sidecar (spec 8.6) was validated on synthetic six-column fixtures only; no real source bytes were opened.",
            "source_tz_offset_minutes production value preserved at 0 from the P3-DE-003 evidence chain (spec Appendix C.2); G-6 requires explicit establishment at source initialization in the pilot phase.",
        ],
    }
    os.makedirs(QUAL_DIR, exist_ok=True)
    with open(os.path.join(QUAL_DIR, "QUALIFICATION_REPORT.json"), "w",
              encoding="utf-8") as fh:
        json.dump(report, fh, indent=2, sort_keys=True)
    _write_md(report)
    return report


def _write_md(report):
    lines = ["# P3-DATA-ENGINE-004 QUALIFICATION REPORT", "",
             "Session: SOURCE-GRAMMAR CORRECTION + RE-QUALIFICATION "
             "(NO REAL-DATA EXECUTION)",
             "Generated: %s" % report["generated_utc"], "",
             "Specification: %s (SHA-256 `%s`)" %
             (report["specification_doc"], report["specification_sha256"]),
             "Governance: %s" % report["governance_baseline"], ""
             "Versions: %s" % ", ".join(
                 "%s=%s" % (k, v) for k, v in report["versions"].items()), "",
             "## Gates", ""]
    for k, g in report["gates"].items():
        lines.append("- %s : %s" % (k, "PASS" if g["pass"] else "FAIL"))
        if k == "source_grammar":
            lines.append("  - suite cases %d, e2e %s"
                         % (g["suite_cases"],
                            "PASS" if g["e2e"] else "FAIL"))
        if k == "mutation":
            lines.append("  - detection %d/%d (per-mutation ledger in JSON)"
                         % (g["detected"], g["total"]))
        if k == "legacy_diff":
            lines.append("  - items classified %d, unresolved %d, deferred %s, "
                         "legacy untouched %s"
                         % (g["items_classified"], g["unresolved"],
                            g["deferred_status"], g["legacy_untouched"]))
    lines += ["", "## Verdict", "", "**%s**" % report["verdict"], "",
              "## Safety", ""]
    for k, v in report["safety"].items():
        lines.append("- %s : %s" % (k, v))
    lines += ["", "## Limitations", ""]
    for lim in report["limitations"]:
        lines.append("- %s" % lim)
    lines += ["", "## Interventions", ""]
    for i, inv in enumerate(report["interventions"], 1):
        lines.append("%d. **%s** -- %s (authorized=%s, %s)"
                     % (i, inv["trigger"], inv["action"],
                        inv["authorization_status"],
                        inv["necessary_or_avoidable"]))
    with open(os.path.join(QUAL_DIR, "QUALIFICATION_REPORT.md"), "w",
              encoding="utf-8") as fh:
        fh.write("\n".join(lines) + "\n")


if __name__ == "__main__":
    report = run_qualification()
    for k, g in report["gates"].items():
        print("%-22s %s" % (k, "PASS" if g["pass"] else "FAIL"))
    print("VERDICT:", report["verdict"])
    sys.exit(0 if report["verdict"] == "QUALIFIED" else 1)