QuantEngineByItz/tools/validate_completion_criteria_v1.py

"""validate_completion_criteria_v1.py — COMPLETION_CRITERIA_VALIDATOR_V1

spec/30_completion_criteria_contract.yaml의 16개 기준을 build_completion_gap_v1.py
산출물(Temp/completion_gap_v1.json)로부터 프로그램적으로 검증한다.

기본 모드: 기준 파일 존재·schema·계산 정확성 검증 (FAIL 기준도 허용).
--require-pass N: N개 이상 PASS 필요.
--strict: 모든 기준 PASS 필요 (이상적 목표 — 현재 달성 불가).

종료코드: 0=검증 통과, 1=검증 실패.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_GAP = ROOT / "Temp" / "completion_gap_v1.json"
SPEC30 = ROOT / "spec" / "30_completion_criteria_contract.yaml"


def _load(path: Path) -> dict:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        print(f"FAIL: cannot load {path}: {e}")
        sys.exit(1)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--gap", default=str(DEFAULT_GAP))
    ap.add_argument("--require-pass", type=int, default=0,
                    help="Required minimum PASS count (0=no requirement)")
    ap.add_argument("--strict", action="store_true",
                    help="All 16 criteria must PASS (ideal goal)")
    args = ap.parse_args()

    gap_path = Path(args.gap) if Path(args.gap).is_absolute() else ROOT / args.gap
    if not gap_path.exists():
        print(f"FAIL: {gap_path} not found - run build-completion-gap-v1 first")
        return 1

    d = _load(gap_path)
    failures: list[str] = []

    # 1) schema 검증
    required = ["formula_id", "total_criteria", "passed_count", "failed_count",
                "pass_rate_pct", "criteria", "priority_roadmap"]
    for f in required:
        if f not in d:
            failures.append(f"missing field: {f}")
    if "workflow_disciplines" not in d:
        failures.append("missing field: workflow_disciplines")

    if failures:
        for f in failures:
            print("FAIL:", f)
        return 1

    total = d["total_criteria"]
    passed = d["passed_count"]
    pass_rate = d["pass_rate_pct"]
    criteria = d["criteria"]

    # 2) 16개 기준 완전성 확인
    expected_criteria_count = 16
    if total != expected_criteria_count:
        failures.append(
            f"total_criteria={total} != {expected_criteria_count} "
            f"— spec/30에 {expected_criteria_count}개 기준 정의됨"
        )

    # 3) 계산 정확성 검증
    actual_passed = sum(1 for c in criteria if c.get("status") == "PASS")
    actual_failed = sum(1 for c in criteria if c.get("status") == "FAIL")
    if actual_passed != passed:
        failures.append(f"passed_count mismatch: declared={passed} actual={actual_passed}")
    expected_rate = round(actual_passed / total * 100, 1) if total else 0
    if abs(expected_rate - pass_rate) > 0.2:
        failures.append(f"pass_rate_pct mismatch: declared={pass_rate} expected={expected_rate}")

    # 4) 각 기준에 필수 필드 존재 확인
    req_crit_fields = ["id", "target", "current", "status", "fix", "effort"]
    for c in criteria:
        for f in req_crit_fields:
            if f not in c:
                failures.append(f"criterion {c.get('id','?')} missing field: {f}")

    # 5) decision_source 필드 확인 (엔진 결정론 기준)
    reproducibility = next(
        (c for c in criteria if c.get("id") == "decision_reproducibility_score"), {}
    )
    if reproducibility.get("status") != "PASS":
        failures.append("CRITICAL: decision_reproducibility_score != PASS — 결정론 미달")

    llm_field = next(
        (c for c in criteria if c.get("id") == "llm_generated_decision_field_count"), {}
    )
    if str(llm_field.get("current")) != "0" and llm_field.get("current") != 0:
        failures.append("CRITICAL: llm_generated_decision_field_count != 0 — LLM 판단 개입")

    workflow = d.get("workflow_disciplines") if isinstance(d.get("workflow_disciplines"), dict) else {}
    required_order = workflow.get("required_preimplementation_order") if isinstance(workflow.get("required_preimplementation_order"), list) else []
    expected_order = [
        "로드맵/현황 확인",
        "WBS 작성",
        "목표 설정",
        "성공판단 데이터 정의",
        "구현",
        "사후 검증",
        "증빙 기록",
    ]
    if required_order != expected_order:
        failures.append(f"workflow_disciplines.required_preimplementation_order mismatch: {required_order}")
    for key in ("completion_gate_rule", "small_change_rule", "scope_change_rule", "evidence_rule"):
        if not str(workflow.get(key) or "").strip():
            failures.append(f"workflow_disciplines missing or empty: {key}")

    if failures:
        for f in failures:
            print("FAIL:", f)
        print(f"COMPLETION_CRITERIA_VALIDATOR_V1: FAIL ({len(failures)} issue(s))")
        return 1

    # 6) --require-pass 검사
    if args.require_pass > 0 and passed < args.require_pass:
        print(
            f"REQUIRE_PASS_FAIL: {passed}/{total} < required {args.require_pass} "
            f"({pass_rate}%)"
        )
        return 1

    # 7) --strict 검사
    if args.strict and passed < total:
        fail_ids = [c["id"] for c in criteria if c.get("status") == "FAIL"]
        print(f"STRICT_FAIL: {total-passed}/{total} criteria still FAIL: {fail_ids}")
        return 1

    print(
        f"COMPLETION_CRITERIA_VALIDATOR_V1: OK | "
        f"{passed}/{total} PASS ({pass_rate}%) | "
        f"deterministic=PASS | llm_fields=0"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())