QuantEngineByItz/tools/validate_gitea_ci_workflow_lint_v1.py

from __future__ import annotations

import argparse
import json
from pathlib import Path

import yaml


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_WORKFLOW = ROOT / ".gitea" / "workflows" / "ci.yml"


def _load_yaml(path: Path) -> dict:
    return yaml.safe_load(path.read_text(encoding="utf-8")) or {}


def main() -> int:
    ap = argparse.ArgumentParser(description="Lint the QuantEngine CI workflow for recurring service-binding mistakes.")
    ap.add_argument("--workflow", default=str(DEFAULT_WORKFLOW))
    args = ap.parse_args()

    workflow_path = Path(args.workflow)
    if not workflow_path.is_absolute():
        workflow_path = (ROOT / workflow_path).resolve()
    data = _load_yaml(workflow_path)

    errors: list[str] = []
    evidence: dict[str, object] = {
        "workflow": str(workflow_path.relative_to(ROOT)),
        "jobs": sorted((data.get("jobs") or {}).keys()),
    }

    jobs = data.get("jobs") or {}
    core = jobs.get("validate-core") or {}
    services = core.get("services") or {}
    postgres = services.get("postgres") or {}
    ci_text = workflow_path.read_text(encoding="utf-8")

    ports = postgres.get("ports") or []
    if any(str(port).strip() == "5432:5432" for port in ports):
        errors.append("validate-core.services.postgres.ports contains fixed host mapping 5432:5432")

    if "PGHOST: postgres" not in ci_text:
        errors.append("workflow does not pin PGHOST=postgres for CI database steps")

    if "QE_WBS_PG_DSN:" not in ci_text or "host=postgres" not in ci_text:
        errors.append("workflow does not publish QE_WBS_PG_DSN with service hostname")

    spec_path = ROOT / "spec" / "60_quant_engine_wbs.yaml"
    spec = _load_yaml(spec_path)
    done_tasks: list[str] = []
    for task_id, task in (spec.get("tasks") or {}).items():
        if (task or {}).get("status") != "DONE":
            continue
        mode = ((task or {}).get("execution") or {}).get("mode")
        if mode == "not_ci_reproducible":
            continue
        done_tasks.append(task_id)
    done_tasks.sort()

    if 'root / "spec" / "60_quant_engine_wbs.yaml"' not in ci_text:
        errors.append("workflow does not derive DONE verdict tasks from spec/60_quant_engine_wbs.yaml")
    if 'mode in {"not_ci_reproducible", "manual_user_action"}' not in ci_text:
        errors.append("workflow does not skip not_ci_reproducible/manual_user_action tasks")
    if "python3 - <<'PY'" not in ci_text:
        errors.append("workflow still hardcodes DONE verdict task list")

    result = {
        "formula_id": "GITEA_CI_WORKFLOW_LINT_V1",
        "gate": "PASS" if not errors else "FAIL",
        "errors": errors,
        "evidence": evidence,
    }
    out = ROOT / "Temp" / "gitea_ci_workflow_lint_v1.json"
    out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if not errors else 1


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