QuantEngineByItz/tools/validate_artifact_sync_v1.py

"""validate_artifact_sync_v1.py — P4-T04: Artifact Sync Validation

Checks that engine_harness_gate_result.json (runtime log) and
formula_runtime_registry_v1.json (artifact file) report consistent gates.
Blocks release if log says PASS/OK but artifact file says FAIL.

formula_id: VALIDATE_ARTIFACT_SYNC_V1
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[1]

DEFAULT_ENGINE_RESULT = ROOT / "Temp" / "engine_harness_gate_result.json"
DEFAULT_REGISTRY     = ROOT / "Temp" / "formula_runtime_registry_v1.json"
DEFAULT_MANIFEST     = ROOT / "runtime" / "active_artifact_manifest.yaml"


def _load_json(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _load_yaml(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        obj = yaml.safe_load(path.read_text(encoding="utf-8"))
        return obj if isinstance(obj, dict) else {}
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _check_engine_result(engine: dict) -> tuple[bool, str]:
    if engine.get("_missing"):
        return False, f"engine_harness_gate_result.json missing: {engine.get('_path')}"
    status = str(engine.get("status") or "").upper()
    # gate field might not exist; status=OK is the success signal
    if status not in ("OK", "PASS"):
        return False, f"engine_harness_gate_result status={status!r} (expected OK/PASS)"
    return True, "OK"


def _check_registry(registry: dict) -> tuple[bool, str]:
    if registry.get("_missing"):
        return False, f"formula_runtime_registry_v1.json missing: {registry.get('_path')}"
    gate = str(registry.get("gate") or "").upper()
    unmapped = registry.get("unmapped_formula_count", -1)
    coverage = registry.get("runtime_adjusted_coverage_pct", 0)
    if gate != "PASS":
        return False, (
            f"formula_runtime_registry gate={gate!r}, "
            f"unmapped={unmapped}, coverage={coverage}%"
        )
    return True, f"gate=PASS coverage={coverage}%"


def _check_manifest(manifest: dict) -> tuple[bool, str]:
    if manifest.get("_missing"):
        return False, f"active_artifact_manifest.yaml missing: {manifest.get('_path')}"
    artifacts = manifest.get("artifacts") or []
    if not artifacts:
        # Some manifests use different structure
        artifacts = list(manifest.values()) if isinstance(manifest, dict) else []

    collision_count = 0
    stale_count = 0
    for art in artifacts:
        if not isinstance(art, dict):
            continue
        if art.get("stale"):
            stale_count += 1
        active_aliases = art.get("active_aliases") or []
        if len(active_aliases) > 1:
            collision_count += 1

    if collision_count > 0:
        return False, f"authority_collision_count={collision_count}"
    if stale_count > 0:
        return False, f"stale_artifact_count={stale_count}"
    return True, "OK"


def _check_sync_consistency(engine_ok: bool, registry_ok: bool) -> tuple[bool, str]:
    """Log PASS but artifact FAIL → ARTIFACT_SYNC_FAIL"""
    if engine_ok and not registry_ok:
        return False, "ARTIFACT_SYNC_MISMATCH: engine_log=OK but registry=FAIL"
    return True, "consistent"


def main() -> int:
    ap = argparse.ArgumentParser(description="P4-T04 artifact sync validator")
    ap.add_argument("--engine-result", default=str(DEFAULT_ENGINE_RESULT))
    ap.add_argument("--manifest",      default=str(DEFAULT_MANIFEST))
    ap.add_argument("--registry",      default=str(DEFAULT_REGISTRY))
    args = ap.parse_args()

    engine   = _load_json(Path(args.engine_result))
    registry = _load_json(Path(args.registry))
    manifest = _load_yaml(Path(args.manifest))

    engine_ok,   engine_msg   = _check_engine_result(engine)
    registry_ok, registry_msg = _check_registry(registry)
    manifest_ok, manifest_msg = _check_manifest(manifest)
    sync_ok,     sync_msg     = _check_sync_consistency(engine_ok, registry_ok)

    mismatches = []
    if not engine_ok:
        mismatches.append(f"engine: {engine_msg}")
    if not registry_ok:
        mismatches.append(f"registry: {registry_msg}")
    if not manifest_ok:
        mismatches.append(f"manifest: {manifest_msg}")
    if not sync_ok:
        mismatches.append(f"sync: {sync_msg}")

    gate = "PASS" if not mismatches else "FAIL"

    result = {
        "formula_id": "VALIDATE_ARTIFACT_SYNC_V1",
        "artifact_sync_mismatch_count": len(mismatches),
        "checks": {
            "engine_result": {"ok": engine_ok, "detail": engine_msg},
            "formula_registry": {"ok": registry_ok, "detail": registry_msg},
            "active_manifest": {"ok": manifest_ok, "detail": manifest_msg},
            "sync_consistency": {"ok": sync_ok, "detail": sync_msg},
        },
        "mismatches": mismatches,
        "gate": gate,
    }

    print(json.dumps(result, ensure_ascii=False, indent=2))

    if gate == "PASS":
        print("VALIDATE_ARTIFACT_SYNC_V1_OK")
        return 0
    else:
        print("VALIDATE_ARTIFACT_SYNC_V1_FAIL")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())