#!/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 evidence_checks = task.get("evidence_checks", []) if not evidence_checks: missing_criteria.append(f"{task_id}.evidence_checks (empty)") else: 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())