feat: add quant engine WBS verification harness
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user