QuantEngineByItz/tools/validate_quant_engine_wbs_v1.py

#!/usr/bin/env python3
from __future__ import annotations

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

import yaml


def load_spec(spec_path: Path) -> dict[str, Any]:
    """Load and parse the WBS spec YAML."""
    if not spec_path.exists():
        raise FileNotFoundError(f"Spec file not found: {spec_path}")
    return yaml.safe_load(spec_path.read_text(encoding="utf-8"))


def read_text(path: Path) -> str:
    """Read text file safely."""
    if not path.exists():
        return ""
    return path.read_text(encoding="utf-8", errors="replace")


def main(argv: list[str] | None = None) -> int:
    """Main entry point."""
    parser = argparse.ArgumentParser(description="Validate whole QuantEngine WBS")
    parser.add_argument("--repo-root", default=None, help="Repository root path")
    parser.add_argument("--spec", default=None, help="Spec YAML file path")

    args = parser.parse_args(argv)

    # Resolve root
    if args.repo_root:
        root = Path(args.repo_root).resolve()
    else:
        root = Path(__file__).resolve().parents[1]

    # Resolve spec path
    if args.spec:
        spec_path = Path(args.spec).resolve()
    else:
        spec_path = root / "spec" / "60_quant_engine_wbs.yaml"

    # Load spec
    try:
        spec = load_spec(spec_path)
    except FileNotFoundError as e:
        print(f"ERROR: {e}")
        return 1

    # Validate spec structure
    meta = spec.get("meta", {})
    formula_id = meta.get("formula_id", "")
    tasks = spec.get("tasks", {})

    missing_criteria = []
    failure_notes = []

    # Check formula ID
    if formula_id != "QUANT_ENGINE_WBS_V1":
        missing_criteria.append("meta.formula_id != QUANT_ENGINE_WBS_V1")
        failure_notes.append("Spec formula_id must be QUANT_ENGINE_WBS_V1")

    # Validate task structure
    valid_statuses = {"PENDING", "IN_PROGRESS", "DONE"}
    for task_id, task in tasks.items():
        # Check required fields
        if "title" not in task:
            missing_criteria.append(f"{task_id}.title")
        if "status" not in task:
            missing_criteria.append(f"{task_id}.status")
        elif task["status"] not in valid_statuses:
            missing_criteria.append(f"{task_id}.status={task['status']}")

        # Check dependencies exist
        depends_on = task.get("depends_on", [])
        for dep_id in depends_on:
            if dep_id not in tasks:
                missing_criteria.append(f"{task_id}.depends_on={dep_id} (task not found)")

        # Check success_criteria structure
        success_criteria = task.get("success_criteria", {})
        if not success_criteria:
            missing_criteria.append(f"{task_id}.success_criteria (missing)")
        else:
            if "expected_success_value" not in success_criteria:
                missing_criteria.append(f"{task_id}.success_criteria.expected_success_value")
            if "evidence_artifacts" not in success_criteria:
                missing_criteria.append(f"{task_id}.success_criteria.evidence_artifacts")
            if "verification_commands" not in success_criteria:
                missing_criteria.append(f"{task_id}.success_criteria.verification_commands")

        # Check evidence_checks (exempt manual_user_action tasks — agents cannot
        # trigger Gitea Actions workflow_dispatch, so there is nothing to gate)
        execution_mode = (task.get("execution") or {}).get("mode")
        evidence_checks = task.get("evidence_checks", [])
        if not evidence_checks and execution_mode != "manual_user_action":
            missing_criteria.append(f"{task_id}.evidence_checks (empty)")
        elif evidence_checks:
            valid_check_types = {"pg_query", "log_pattern", "json_gate", "file_exists", "playwright_report"}
            for check in evidence_checks:
                check_type = check.get("type", "")
                if check_type not in valid_check_types:
                    missing_criteria.append(f"{task_id}.evidence_checks[] type={check_type} (unknown)")

    # Check DONE tasks have verdicts
    per_task = {}
    for task_id, task in tasks.items():
        status = task.get("status", "")
        verdict_path = root / "Temp" / "evidence" / task_id / "verdict.json"
        verdict_gate = None

        if status == "DONE":
            if not verdict_path.exists():
                missing_criteria.append(f"{task_id}.verdict (missing for DONE task)")
                failure_notes.append(
                    f"Task {task_id} has status=DONE but verdict.json is missing. "
                    f"Run: python tools/verify_wbs_task_v1.py --task {task_id}"
                )
            else:
                try:
                    verdict = json.loads(verdict_path.read_text(encoding="utf-8"))
                    verdict_gate = verdict.get("gate", "")
                    if verdict_gate != "PASS":
                        missing_criteria.append(f"{task_id}.verdict gate={verdict_gate} (not PASS)")
                        failure_notes.append(
                            f"Task {task_id} has status=DONE but verdict.json gate={verdict_gate}. "
                            f"Fix evidence and re-run: python tools/verify_wbs_task_v1.py --task {task_id}"
                        )
                except Exception as e:
                    missing_criteria.append(f"{task_id}.verdict (parse error: {e})")
                    failure_notes.append(f"Task {task_id} verdict.json is invalid: {e}")

            # Check dependencies are DONE
            depends_on = task.get("depends_on", [])
            for dep_id in depends_on:
                dep_task = tasks.get(dep_id, {})
                dep_status = dep_task.get("status", "")
                if dep_status != "DONE":
                    missing_criteria.append(f"{task_id}.depends_on={dep_id} (not DONE, is {dep_status})")
                    failure_notes.append(
                        f"Task {task_id} depends on {dep_id}, but {dep_id} status={dep_status} (not DONE)"
                    )

        per_task[task_id] = {
            "status": status,
            "verdict_gate": verdict_gate
        }

    # Check roadmap doc pointer
    roadmap_path = root / "docs" / "ROADMAP_WBS.md"
    if not roadmap_path.exists():
        missing_criteria.append("docs/ROADMAP_WBS.md (missing)")
        failure_notes.append("Roadmap document is missing at docs/ROADMAP_WBS.md")
    else:
        roadmap_text = read_text(roadmap_path)
        if "QUANT_ENGINE_WBS_V1" not in roadmap_text:
            missing_criteria.append("docs/ROADMAP_WBS.md (no QUANT_ENGINE_WBS_V1 reference)")
            failure_notes.append(
                "Roadmap document does not contain 'QUANT_ENGINE_WBS_V1' reference. "
                "Add a pointer to docs/ROADMAP_WBS.md"
            )

    # Build result
    gate = "PASS" if not missing_criteria else "FAIL"
    message = (
        "QuantEngine WBS validation passed."
        if gate == "PASS"
        else "QuantEngine WBS validation failed. See failure_notes for details."
    )

    payload = {
        "formula_id": "QUANT_ENGINE_WBS_V1",
        "gate": gate,
        "message": message,
        "spec_path": str(spec_path),
        "task_count": len(tasks),
        "done_count": sum(1 for t in tasks.values() if t.get("status") == "DONE"),
        "missing_criteria": missing_criteria,
        "failure_notes": failure_notes,
        "per_task": per_task
    }

    # Save output
    out_path = root / "Temp" / "quant_engine_wbs_v1.json"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")

    # Print result
    print(message)
    print(json.dumps(payload, ensure_ascii=False, indent=2))

    return 0 if gate == "PASS" else 1


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