QuantEngineByItz/tools/validate_pipeline_runtime_contract.py

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any

import yaml


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = ROOT / "spec" / "22_pipeline_runtime_contract.yaml"
DEFAULT_PROFILE = ROOT / "Temp" / "pipeline_runtime_profile_v1.json"
DEFAULT_GATE = ROOT / "Temp" / "engine_harness_gate_result.json"


def _load_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        obj = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def _load_yaml(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        obj = yaml.safe_load(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--contract", default=str(DEFAULT_CONTRACT))
    ap.add_argument("--profile", default=str(DEFAULT_PROFILE))
    ap.add_argument("--gate", default=str(DEFAULT_GATE))
    args = ap.parse_args()

    cp = Path(args.contract)
    pp = Path(args.profile)
    gp = Path(args.gate)
    if not cp.is_absolute():
        cp = ROOT / cp
    if not pp.is_absolute():
        pp = ROOT / pp
    if not gp.is_absolute():
        gp = ROOT / gp

    contract = _load_yaml(cp).get("pipeline_runtime_contract", {})
    profile = _load_json(pp)
    gate = _load_json(gp)

    mode = str(profile.get("mode") or "")
    modes = contract.get("modes") if isinstance(contract.get("modes"), dict) else {}
    mode_cfg = modes.get(mode) if isinstance(modes.get(mode), dict) else {}
    max_target = float(mode_cfg.get("max_elapsed_sec_target") or 0.0)
    elapsed = float(profile.get("elapsed_sec_total") or 0.0)
    dup_removed = int(profile.get("duplicate_steps_removed_count") or 0)
    steps = profile.get("steps") if isinstance(profile.get("steps"), list) else []

    runtime_ctx = profile.get("runtime_context") if isinstance(profile.get("runtime_context"), dict) else {}
    skip_validate = bool(runtime_ctx.get("skip_validate") if runtime_ctx.get("skip_validate") is not None else profile.get("skip_validate"))
    allowed_use = str(profile.get("allowed_use") or "")

    failed: list[str] = []
    warnings: list[str] = []
    if not mode_cfg:
        failed.append("MODE_NOT_IN_CONTRACT")
    if max_target > 0 and elapsed > max_target:
        warnings.append(f"ELAPSED_OVER_TARGET:{elapsed}>{max_target}")
    if str(gate.get("status") or "") != "OK":
        failed.append("ENGINE_GATE_NOT_OK")
    # failed_checks 중 warn_only가 아닌 hard-fail만 체크
    all_checks = gate.get("checks") if isinstance(gate.get("checks"), list) else []
    hard_fails = [
        c.get("name") for c in all_checks
        if isinstance(c, dict) and c.get("exit_code", 0) != 0 and not c.get("warn_only", False)
    ]
    if hard_fails:
        failed.append("ENGINE_GATE_HARD_FAIL_EXISTS")
    if mode == "release" and dup_removed < 1:
        warnings.append("DUPLICATE_STEPS_NOT_REMOVED")
    if len(steps) == 0 and mode != "package-only":
        failed.append("PROFILE_STEPS_EMPTY")

    if mode == "release" and skip_validate:
        failed.append("RELEASE_MODE_SKIP_VALIDATE_NOT_ALLOWED")

    expected_allowed_use = "production_investment_decisions" if mode in {"release", "quick"} else "packaging_only"
    if mode_cfg and allowed_use != expected_allowed_use:
        failed.append("ALLOWED_USE_MISMATCH")

    release_mode_skip_validate_count = 1 if (mode == "release" and skip_validate) else 0
    package_only_used_for_investment_decision_count = 1 if (mode == "package-only" and allowed_use == "production_investment_decisions") else 0

    status = "FAIL" if failed else "OK"
    result = {
        "formula_id": "PIPELINE_RUNTIME_CONTRACT_VALIDATOR_V1",
        "status": status,
        "mode": mode,
        "elapsed_sec_total": elapsed,
        "max_elapsed_sec_target": max_target,
        "release_mode_skip_validate_count": release_mode_skip_validate_count,
        "package_only_used_for_investment_decision_count": package_only_used_for_investment_decision_count,
        "failed": failed,
        "warnings": warnings,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 1 if failed else 0


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