"""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)