208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
#!/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
|
|
|
|
execution_mode = (task.get("execution") or {}).get("mode")
|
|
if status == "DONE" and execution_mode == "not_ci_reproducible":
|
|
# Evidence requires a live app/KIS API round-trip that a stateless CI runner
|
|
# cannot reproduce on demand (e.g. Hangfire job + real collection run). The
|
|
# task was verified DONE against real local data; CI trusts that record
|
|
# without re-deriving it (see spec/60 meta.execution_convention).
|
|
pass
|
|
elif 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}")
|
|
|
|
if status == "DONE":
|
|
# Check dependencies are DONE (applies regardless of execution_mode)
|
|
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())
|