| """External completion gate for the full original DataForge vision.
|
|
|
| This gate is intentionally stricter than the local release gate. It checks
|
| public package names, the production Workers playground, Hugging Face artifacts,
|
| and evidence files that can only be produced by real external work.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import sys
|
| import urllib.error
|
| import urllib.request
|
| from dataclasses import asdict, dataclass
|
| from pathlib import Path
|
| from typing import Any
|
|
|
| from dataforge.release.model_family import (
|
| FAMILY_REPORT_SCHEMA_VERSION,
|
| MODEL_FAMILY_SIZES,
|
| MODEL_FAMILY_STAGES,
|
| PASSING_QUALITY_STATUSES,
|
| PUBLIC_ARTIFACT_STATUSES,
|
| license_matches,
|
| load_model_family_manifest,
|
| )
|
|
|
| SCHEMA_VERSION = "dataforge_full_vision_gate_v1"
|
| EXPECTED_VERSION = "0.1.0"
|
| EXPECTED_PACKAGES = (
|
| "dataforge_07",
|
| "dataforge_07_mcp",
|
| "dataforge_07_evals",
|
| "dataforge_07_dbt",
|
| "dataforge_07_agent_patterns",
|
| )
|
| EXPECTED_MODEL_SIZES = MODEL_FAMILY_SIZES
|
| EXPECTED_MODEL_STAGES = MODEL_FAMILY_STAGES
|
| EXPECTED_DESIGN_PERSONAS = ("marcus", "priya", "shreya", "agent")
|
| DEFAULT_FRONTEND_URL = "https://dataforge.praneshrajan15.workers.dev/playground"
|
| DEFAULT_BACKEND_URL = "https://Praneshrajan15-dataforge-playground.hf.space"
|
| DEFAULT_HF_OWNER = "Praneshrajan15"
|
| REQUIRED_PACKAGE_EVIDENCE_FIELDS = (
|
| "version",
|
| "workflow_run_url",
|
| "testpypi_smoke_log_path",
|
| "pypi_smoke_log_path",
|
| )
|
| EXPECTED_PUBLISHER_REPOSITORY = "Aegis15/dataforge"
|
| EXPECTED_OIDC_ISSUER = "https://token.actions.githubusercontent.com"
|
| PUBLISH_ATTESTATION_PREDICATE = "https://docs.pypi.org/attestations/publish/v1"
|
| PYPI_WORKFLOWS = {
|
| "dataforge_07": "publish-dataforge.yml",
|
| "dataforge_07_mcp": "publish-dataforge-mcp.yml",
|
| "dataforge_07_evals": "publish-dataforge-evals.yml",
|
| "dataforge_07_dbt": "publish-dataforge-dbt.yml",
|
| "dataforge_07_agent_patterns": "publish-dataforge-agent-patterns.yml",
|
| }
|
| TESTPYPI_WORKFLOWS = {
|
| "dataforge_07": "publish-testpypi.yml",
|
| "dataforge_07_mcp": "publish-dataforge-mcp-testpypi.yml",
|
| "dataforge_07_evals": "publish-dataforge-evals-testpypi.yml",
|
| "dataforge_07_dbt": "publish-dataforge-dbt-testpypi.yml",
|
| "dataforge_07_agent_patterns": "publish-dataforge-agent-patterns-testpypi.yml",
|
| }
|
|
|
|
|
| @dataclass(frozen=True)
|
| class FullVisionCheck:
|
| """One full-vision completion check."""
|
|
|
| name: str
|
| ok: bool
|
| detail: str
|
| metadata: dict[str, Any]
|
|
|
|
|
| @dataclass(frozen=True)
|
| class FullVisionReport:
|
| """Machine-readable full-vision completion report."""
|
|
|
| ok: bool
|
| checks: list[FullVisionCheck]
|
| schema_version: str = SCHEMA_VERSION
|
|
|
| def to_dict(self) -> dict[str, Any]:
|
| """Return a JSON-serializable report."""
|
| return asdict(self)
|
|
|
|
|
| def _project_root() -> Path:
|
| """Return the repository root."""
|
| return Path(__file__).resolve().parents[2]
|
|
|
|
|
| def _evidence_root(root: Path | None = None) -> Path:
|
| """Return the release evidence directory."""
|
| return root if root is not None else _project_root() / "docs" / "evidence"
|
|
|
|
|
| def _fetch_json(url: str, *, timeout_s: float = 20.0) -> tuple[dict[str, Any] | None, str]:
|
| """Fetch JSON from a public endpoint."""
|
| request = urllib.request.Request(url, headers={"User-Agent": "dataforge-full-vision-gate"})
|
| try:
|
| with urllib.request.urlopen(request, timeout=timeout_s) as response:
|
| return json.loads(response.read().decode("utf-8")), ""
|
| except urllib.error.HTTPError as exc:
|
| return None, f"HTTP {exc.code}"
|
| except Exception as exc:
|
| return None, str(exc)
|
|
|
|
|
| def _fetch_text(
|
| url: str,
|
| *,
|
| timeout_s: float = 20.0,
|
| headers: dict[str, str] | None = None,
|
| ) -> tuple[int | None, str, dict[str, str], str]:
|
| """Fetch text and return status, body, response headers, and error."""
|
| request = urllib.request.Request(
|
| url,
|
| headers={"User-Agent": "dataforge-full-vision-gate", **(headers or {})},
|
| )
|
| try:
|
| with urllib.request.urlopen(request, timeout=timeout_s) as response:
|
| response_headers = {key.lower(): value for key, value in response.headers.items()}
|
| return (
|
| int(response.status),
|
| response.read().decode("utf-8", errors="replace"),
|
| response_headers,
|
| "",
|
| )
|
| except urllib.error.HTTPError as exc:
|
| return int(exc.code), exc.read().decode("utf-8", errors="replace"), {}, f"HTTP {exc.code}"
|
| except Exception as exc:
|
| return None, "", {}, str(exc)
|
|
|
|
|
| def _load_json_file(path: Path) -> tuple[dict[str, Any] | None, str]:
|
| """Load a JSON evidence file."""
|
| if not path.exists():
|
| return None, f"Missing evidence file: {path}"
|
| try:
|
| payload = json.loads(path.read_text(encoding="utf-8"))
|
| except Exception as exc:
|
| return None, f"Invalid JSON evidence file {path}: {exc}"
|
| if not isinstance(payload, dict):
|
| return None, f"Evidence file {path} must contain a JSON object."
|
| return payload, ""
|
|
|
|
|
| def _resolve_evidence_file(evidence_root: Path, raw_path: Any) -> Path | None:
|
| """Resolve a manifest evidence path and require it to point to a file."""
|
| text = str(raw_path or "").strip()
|
| if not text:
|
| return None
|
| path = Path(text)
|
| if not path.is_absolute():
|
| path = evidence_root / path
|
| if not path.is_file():
|
| return None
|
| return path
|
|
|
|
|
| def _is_https_url(value: Any) -> bool:
|
| """Return whether a manifest field is a concrete HTTPS URL."""
|
| return str(value or "").strip().startswith("https://")
|
|
|
|
|
| def _is_sha256(value: Any) -> bool:
|
| """Return whether a manifest field looks like a SHA-256 hex digest."""
|
| text = str(value or "").strip().lower()
|
| return len(text) == 64 and all(char in "0123456789abcdef" for char in text)
|
|
|
|
|
| def _check_distribution_evidence(
|
| *,
|
| package: str,
|
| index: str,
|
| kind: str,
|
| payload: Any,
|
| expected_workflow: str,
|
| ) -> list[str]:
|
| """Return errors for one wheel or sdist evidence object."""
|
| errors: list[str] = []
|
| label = f"{package}.{index}.{kind}"
|
| if not isinstance(payload, dict):
|
| return [f"{label}: evidence must be an object"]
|
| for field in (
|
| "filename",
|
| "package_type",
|
| "download_url",
|
| "sha256",
|
| "upload_time_iso_8601",
|
| "provenance_url",
|
| "integrity_predicate_type",
|
| "integrity_subject_sha256",
|
| ):
|
| if not str(payload.get(field, "")).strip():
|
| errors.append(f"{label}: {field} is required")
|
| for field in ("download_url", "provenance_url"):
|
| if payload.get(field) and not _is_https_url(payload.get(field)):
|
| errors.append(f"{label}: {field} must be an HTTPS URL")
|
| sha256 = str(payload.get("sha256", "")).lower()
|
| subject_sha256 = str(payload.get("integrity_subject_sha256", "")).lower()
|
| if sha256 and not _is_sha256(sha256):
|
| errors.append(f"{label}: sha256 must be a SHA-256 hex digest")
|
| if subject_sha256 and not _is_sha256(subject_sha256):
|
| errors.append(f"{label}: integrity_subject_sha256 must be a SHA-256 hex digest")
|
| if sha256 and subject_sha256 and sha256 != subject_sha256:
|
| errors.append(f"{label}: integrity_subject_sha256 must match sha256")
|
| if payload.get("integrity_predicate_type") != PUBLISH_ATTESTATION_PREDICATE:
|
| errors.append(f"{label}: integrity_predicate_type must be {PUBLISH_ATTESTATION_PREDICATE}")
|
| publisher = payload.get("trusted_publisher")
|
| if not isinstance(publisher, dict):
|
| errors.append(f"{label}: trusted_publisher must be an object")
|
| return errors
|
| if publisher.get("repository") != EXPECTED_PUBLISHER_REPOSITORY:
|
| errors.append(
|
| f"{label}: trusted_publisher.repository must be {EXPECTED_PUBLISHER_REPOSITORY}"
|
| )
|
| if publisher.get("workflow") != expected_workflow:
|
| errors.append(f"{label}: trusted_publisher.workflow must be {expected_workflow}")
|
| if publisher.get("oidc_issuer") != EXPECTED_OIDC_ISSUER:
|
| errors.append(f"{label}: trusted_publisher.oidc_issuer must be {EXPECTED_OIDC_ISSUER}")
|
| identity = str(publisher.get("identity", ""))
|
| if identity and expected_workflow not in identity:
|
| errors.append(f"{label}: trusted_publisher.identity must reference {expected_workflow}")
|
| return errors
|
|
|
|
|
| def _check_index_evidence(
|
| *,
|
| package: str,
|
| index: str,
|
| payload: Any,
|
| expected_workflow: str,
|
| ) -> list[str]:
|
| """Return errors for one package-index evidence object."""
|
| errors: list[str] = []
|
| label = f"{package}.{index}"
|
| if not isinstance(payload, dict):
|
| return [f"{label}: evidence must be an object"]
|
| if payload.get("index") != index:
|
| errors.append(f"{label}: index must be {index}")
|
| if not _is_https_url(payload.get("project_url")):
|
| errors.append(f"{label}: project_url must be an HTTPS URL")
|
| for kind in ("wheel", "sdist"):
|
| errors.extend(
|
| _check_distribution_evidence(
|
| package=package,
|
| index=index,
|
| kind=kind,
|
| payload=payload.get(kind),
|
| expected_workflow=expected_workflow,
|
| )
|
| )
|
| return errors
|
|
|
|
|
| def _version_tuple(value: str) -> tuple[int, ...]:
|
| """Parse a simple dotted numeric version for release-gate comparisons."""
|
| parts: list[int] = []
|
| for raw_part in value.split("."):
|
| digits = "".join(ch for ch in raw_part if ch.isdigit())
|
| parts.append(int(digits or "0"))
|
| return tuple(parts)
|
|
|
|
|
| def _check_package_index(index_name: str, base_url: str) -> FullVisionCheck:
|
| """Verify all final package names exist on one package index."""
|
| package_results: dict[str, Any] = {}
|
| errors: list[str] = []
|
| for package in EXPECTED_PACKAGES:
|
| url = f"{base_url.rstrip('/')}/{package}/json"
|
| payload, error = _fetch_json(url)
|
| if payload is None:
|
| package_results[package] = {"ok": False, "error": error}
|
| errors.append(f"{package}: {error}")
|
| continue
|
| info = payload.get("info", {}) if isinstance(payload.get("info"), dict) else {}
|
| version = str(info.get("version", ""))
|
| summary = str(info.get("summary", ""))
|
| home_page = str(info.get("home_page", ""))
|
| project_urls = info.get("project_urls", {})
|
| project_urls_text = (
|
| json.dumps(project_urls, sort_keys=True) if isinstance(project_urls, dict) else ""
|
| )
|
| ok = _version_tuple(version) >= _version_tuple(EXPECTED_VERSION)
|
| if package == "dataforge":
|
| metadata_text = f"{summary} {home_page} {project_urls_text}".lower()
|
| ok = (
|
| ok
|
| and ("dataforge" in metadata_text)
|
| and ("aegis15" in metadata_text or "data-quality" in metadata_text)
|
| )
|
| package_results[package] = {
|
| "ok": ok,
|
| "version": version,
|
| "summary": summary,
|
| "home_page": home_page,
|
| "project_urls": project_urls,
|
| "url": url,
|
| }
|
| if not ok:
|
| errors.append(
|
| f"{package}: found version={version!r}, summary={summary!r}, home_page={home_page!r}"
|
| )
|
| return FullVisionCheck(
|
| name=f"{index_name}_packages",
|
| ok=not errors,
|
| detail=f"All final DataForge packages are published on {index_name}."
|
| if not errors
|
| else f"{index_name} package publication is incomplete or not owned by this project.",
|
| metadata={"packages": package_results, "errors": errors},
|
| )
|
|
|
|
|
| def _check_publish_evidence(evidence_root: Path) -> FullVisionCheck:
|
| """Verify local evidence for trusted publishing and attestations."""
|
| path = evidence_root / "pypi" / "publish_report.json"
|
| payload, error = _load_json_file(path)
|
| if payload is None:
|
| return FullVisionCheck("pypi_publish_evidence", False, error, {"path": str(path)})
|
| packages = payload.get("packages")
|
| errors: list[str] = []
|
| if payload.get("schema_version") != "dataforge_pypi_publish_report_v2":
|
| errors.append("schema_version must be dataforge_pypi_publish_report_v2")
|
| if not isinstance(packages, list):
|
| errors.append("packages must be a list")
|
| packages = []
|
| seen = {str(item.get("name", "")) for item in packages if isinstance(item, dict)}
|
| missing = sorted(set(EXPECTED_PACKAGES) - seen)
|
| errors.extend(f"missing package evidence: {name}" for name in missing)
|
| for item in packages:
|
| if not isinstance(item, dict):
|
| errors.append("package evidence entries must be objects")
|
| continue
|
| name = str(item.get("name", ""))
|
| for field in REQUIRED_PACKAGE_EVIDENCE_FIELDS:
|
| if not str(item.get(field, "")).strip():
|
| errors.append(f"{name}: {field} is required")
|
| if item.get("version") and str(item.get("version")) != EXPECTED_VERSION:
|
| errors.append(f"{name}: version must be {EXPECTED_VERSION}")
|
| if item.get("trusted_publishing") is not True:
|
| errors.append(f"{name}: trusted_publishing must be true")
|
| if item.get("attestations") is not True:
|
| errors.append(f"{name}: attestations must be true")
|
| if item.get("testpypi_fresh_install") is not True:
|
| errors.append(f"{name}: TestPyPI fresh-install proof is missing")
|
| if item.get("pypi_fresh_install") is not True:
|
| errors.append(f"{name}: PyPI fresh-install proof is missing")
|
| for field in ("workflow_run_url",):
|
| if item.get(field) and not _is_https_url(item.get(field)):
|
| errors.append(f"{name}: {field} must be an HTTPS URL")
|
| for field in ("testpypi_smoke_log_path", "pypi_smoke_log_path"):
|
| if item.get(field) and _resolve_evidence_file(evidence_root, item.get(field)) is None:
|
| errors.append(f"{name}: {field} must point to an evidence file")
|
| if name in PYPI_WORKFLOWS:
|
| errors.extend(
|
| _check_index_evidence(
|
| package=name,
|
| index="pypi",
|
| payload=item.get("pypi"),
|
| expected_workflow=PYPI_WORKFLOWS[name],
|
| )
|
| )
|
| errors.extend(
|
| _check_index_evidence(
|
| package=name,
|
| index="testpypi",
|
| payload=item.get("testpypi"),
|
| expected_workflow=TESTPYPI_WORKFLOWS[name],
|
| )
|
| )
|
| return FullVisionCheck(
|
| name="pypi_publish_evidence",
|
| ok=not errors,
|
| detail="Trusted publishing, attestations, provenance, and fresh-install evidence exist."
|
| if not errors
|
| else "Publishing evidence is incomplete.",
|
| metadata={"path": str(path), "errors": errors},
|
| )
|
|
|
|
|
| def _check_workers_playground(frontend_url: str, backend_url: str) -> FullVisionCheck:
|
| """Verify the production Workers playground serves the frontend shell."""
|
| errors: list[str] = []
|
| metadata: dict[str, Any] = {"frontend_url": frontend_url, "backend_url": backend_url}
|
|
|
| status, body, headers, error = _fetch_text(frontend_url)
|
| metadata["status_code"] = status
|
| metadata["content_type"] = headers.get("content-type", "")
|
| if error:
|
| errors.append(f"frontend fetch failed: {error}")
|
| if status != 200:
|
| errors.append(f"frontend returned {status}, expected 200")
|
| lowered = body.lower()
|
| for marker in ("<!doctype html>", 'id="root"', "config.js"):
|
| if marker not in lowered:
|
| errors.append(f"frontend HTML missing marker {marker!r}")
|
| config_url = f"{frontend_url.rstrip('/')}/config.js"
|
| config_status, config_body, config_headers, config_error = _fetch_text(config_url)
|
| metadata["config_status_code"] = config_status
|
| metadata["config_cache_control"] = config_headers.get("cache-control", "")
|
| if config_error:
|
| errors.append(f"config.js fetch failed: {config_error}")
|
| if backend_url not in config_body:
|
| errors.append("config.js does not point at the expected HF backend")
|
| if "no-store" not in metadata["config_cache_control"].lower():
|
| errors.append("config.js is not served with Cache-Control: no-store")
|
| return FullVisionCheck(
|
| name="workers_dev_playground",
|
| ok=not errors,
|
| detail="workers.dev playground serves the production frontend."
|
| if not errors
|
| else "workers.dev playground is not production-ready.",
|
| metadata={**metadata, "errors": errors},
|
| )
|
|
|
|
|
| def _check_hf_backend(
|
| frontend_url: str, backend_url: str, expected_git_sha: str | None
|
| ) -> FullVisionCheck:
|
| """Verify the deployed HF Space backend and CORS."""
|
| errors: list[str] = []
|
| health_url = f"{backend_url.rstrip('/')}/api/health"
|
| status, body, _headers, error = _fetch_text(health_url)
|
| metadata: dict[str, Any] = {
|
| "backend_url": backend_url,
|
| "health_url": health_url,
|
| "status_code": status,
|
| }
|
| if error:
|
| errors.append(f"health fetch failed: {error}")
|
| payload: dict[str, Any] = {}
|
| if status == 200:
|
| try:
|
| decoded = json.loads(body)
|
| if isinstance(decoded, dict):
|
| payload = decoded
|
| except json.JSONDecodeError as exc:
|
| errors.append(f"health JSON decode failed: {exc}")
|
| else:
|
| errors.append(f"health returned {status}, expected 200")
|
| metadata["health"] = payload
|
| if payload.get("status") != "ok":
|
| errors.append("health status is not ok")
|
| if payload.get("environment") != "production":
|
| errors.append("health environment is not production")
|
| if expected_git_sha:
|
| build_sha = str(payload.get("build_sha", ""))
|
| if not build_sha.startswith(expected_git_sha[:12]):
|
| errors.append(
|
| f"build_sha {build_sha!r} does not match release SHA {expected_git_sha!r}"
|
| )
|
|
|
| origin = frontend_url.split("/playground", 1)[0]
|
| cors_status, _cors_body, cors_headers, cors_error = _fetch_text(
|
| health_url,
|
| headers={"Origin": origin},
|
| )
|
| metadata["cors_status_code"] = cors_status
|
| metadata["cors_allow_origin"] = cors_headers.get("access-control-allow-origin", "")
|
| if cors_error:
|
| errors.append(f"CORS probe failed: {cors_error}")
|
| if metadata["cors_allow_origin"] != origin:
|
| errors.append(f"CORS does not allow frontend origin {origin}")
|
| return FullVisionCheck(
|
| name="hf_space_backend",
|
| ok=not errors,
|
| detail="HF Space backend is live, production, and CORS-compatible."
|
| if not errors
|
| else "HF Space backend verification failed.",
|
| metadata={**metadata, "errors": errors},
|
| )
|
|
|
|
|
| def _check_dbt_evidence(evidence_root: Path) -> FullVisionCheck:
|
| """Verify the dbt-duckdb fresh environment proof."""
|
| path = evidence_root / "dbt_duckdb" / "fresh_env_report.json"
|
| payload, error = _load_json_file(path)
|
| if payload is None:
|
| return FullVisionCheck("dbt_duckdb_fresh_env", False, error, {"path": str(path)})
|
| errors: list[str] = []
|
| if payload.get("schema_version") != "dataforge_dbt_fresh_env_proof_v1":
|
| errors.append("schema_version must be dataforge_dbt_fresh_env_proof_v1")
|
| if payload.get("install_source") != "pypi":
|
| errors.append("install_source must be pypi")
|
| if payload.get("package") != "dataforge_07_dbt":
|
| errors.append("package must be dataforge_07_dbt")
|
| if not str(payload.get("python_version", "")).startswith("3.12"):
|
| errors.append("python_version must be 3.12.x")
|
| if not str(payload.get("dbt_core_version", "")).strip():
|
| errors.append("dbt_core_version is required")
|
| if not str(payload.get("dbt_duckdb_version", "")).strip():
|
| errors.append("dbt_duckdb_version is required")
|
| for field in ("dbt_seed_passed", "dbt_run_passed", "dbt_test_passed"):
|
| if payload.get(field) is not True:
|
| errors.append(f"{field} must be true")
|
| for field in (
|
| "dataforge_dbt_dry_run_passed",
|
| "dataforge_dbt_refuse_passed",
|
| "dataforge_dbt_apply_passed",
|
| "dataforge_table_store_audit_passed",
|
| "dataforge_table_store_revert_passed",
|
| ):
|
| if payload.get(field) is not True:
|
| errors.append(f"{field} must be true")
|
| if payload.get("dbt_duckdb_e2e_passed") is not True:
|
| errors.append("dbt_duckdb_e2e_passed must be true")
|
| if int(payload.get("skipped_tests", -1)) != 0:
|
| errors.append("skipped_tests must be 0")
|
| if payload.get("audit_artifact_written") is not True:
|
| errors.append("audit_artifact_written must be true")
|
| for field in ("artifact_path", "command_log_path"):
|
| if _resolve_evidence_file(evidence_root, payload.get(field)) is None:
|
| errors.append(f"{field} must point to an evidence file")
|
| return FullVisionCheck(
|
| name="dbt_duckdb_fresh_env",
|
| ok=not errors,
|
| detail="dbt-duckdb fresh-env proof is complete."
|
| if not errors
|
| else "dbt-duckdb fresh-env proof is incomplete.",
|
| metadata={"path": str(path), "errors": errors},
|
| )
|
|
|
|
|
| def _check_design_partner_evidence(evidence_root: Path) -> FullVisionCheck:
|
| """Verify real design-partner evidence exists for every required path."""
|
| path = evidence_root / "design_partners" / "manifest.json"
|
| payload, error = _load_json_file(path)
|
| if payload is None:
|
| return FullVisionCheck("design_partner_evidence", False, error, {"path": str(path)})
|
| errors: list[str] = []
|
| if payload.get("schema_version") != "dataforge_design_partner_manifest_v1":
|
| errors.append("schema_version must be dataforge_design_partner_manifest_v1")
|
| entries = payload.get("validations")
|
| if not isinstance(entries, list):
|
| errors.append("validations must be a list")
|
| entries = []
|
| by_persona = {
|
| str(entry.get("persona", "")).lower(): entry for entry in entries if isinstance(entry, dict)
|
| }
|
| for persona in EXPECTED_DESIGN_PERSONAS:
|
| entry = by_persona.get(persona)
|
| if entry is None:
|
| errors.append(f"missing design-partner validation: {persona}")
|
| continue
|
| for field in ("role", "session_date", "production_surface", "task_completed"):
|
| if not str(entry.get(field, "")).strip():
|
| errors.append(f"{persona}: {field} is required")
|
| if entry.get("validated") is not True:
|
| errors.append(f"{persona}: validated must be true")
|
| if entry.get("trust_confirmed") is not True:
|
| errors.append(f"{persona}: trust_confirmed must be true")
|
| if not entry.get("evidence_path"):
|
| errors.append(f"{persona}: evidence_path is required")
|
| elif _resolve_evidence_file(evidence_root, entry.get("evidence_path")) is None:
|
| errors.append(f"{persona}: evidence_path must point to an evidence file")
|
| if not entry.get("consent_record"):
|
| errors.append(f"{persona}: consent_record is required")
|
| elif _resolve_evidence_file(evidence_root, entry.get("consent_record")) is None:
|
| errors.append(f"{persona}: consent_record must point to an evidence file")
|
| if entry.get("blocking_findings_closed") is not True:
|
| errors.append(f"{persona}: blocking_findings_closed must be true")
|
| timing = entry.get("timing_seconds")
|
| if not isinstance(timing, int | float) or timing <= 0:
|
| errors.append(f"{persona}: timing_seconds must be positive")
|
| return FullVisionCheck(
|
| name="design_partner_evidence",
|
| ok=not errors,
|
| detail="Design-partner evidence covers Marcus, Priya, Shreya, and agent paths."
|
| if not errors
|
| else "Design-partner evidence is incomplete.",
|
| metadata={"path": str(path), "errors": errors},
|
| )
|
|
|
|
|
| def _check_hf_model_family(evidence_root: Path, hf_owner: str) -> FullVisionCheck:
|
| """Verify public HF model repos and local quality evidence for the full family."""
|
| errors: list[str] = []
|
| repo_results: dict[str, Any] = {}
|
| manifest = load_model_family_manifest()
|
| expected_entries = tuple(manifest.entries)
|
| expected_repos = {entry.repo_id for entry in expected_entries}
|
| for entry in expected_entries:
|
| repo_id = entry.repo_id
|
| if hf_owner != manifest.hf_owner:
|
| repo_id = f"{hf_owner}/DataForge-{entry.size}-{entry.stage}"
|
| payload, error = _fetch_json(f"https://huggingface.co/api/models/{repo_id}")
|
| if payload is None:
|
| repo_results[repo_id] = {"ok": False, "error": error}
|
| errors.append(f"{repo_id}: {error}")
|
| continue
|
| card = payload.get("cardData", {})
|
| card = card if isinstance(card, dict) else {}
|
| card_errors = _check_model_card_fields(
|
| repo_id=repo_id,
|
| card=card,
|
| expected_base_model=entry.base_model,
|
| expected_license=entry.upstream_license,
|
| expected_dataset=manifest.dataset_repo,
|
| )
|
| ok = not card_errors
|
| repo_results[repo_id] = {
|
| "ok": ok,
|
| "sha": payload.get("sha"),
|
| "lastModified": payload.get("lastModified"),
|
| "card_errors": card_errors,
|
| }
|
| errors.extend(card_errors)
|
|
|
| path = evidence_root / "models" / "model_family_report.json"
|
| payload, error = _load_json_file(path)
|
| if payload is None:
|
| errors.append(error)
|
| evidence_metadata: dict[str, Any] = {"path": str(path), "loaded": False}
|
| else:
|
| evidence_metadata = {"path": str(path), "loaded": True}
|
| if payload.get("schema_version") != FAMILY_REPORT_SCHEMA_VERSION:
|
| errors.append(
|
| f"model_family_report schema_version must be {FAMILY_REPORT_SCHEMA_VERSION}"
|
| )
|
| models = payload.get("models")
|
| if not isinstance(models, list):
|
| errors.append("model_family_report models must be a list")
|
| models = []
|
| entries_by_repo = {
|
| str(item.get("repo_id", "")): item for item in models if isinstance(item, dict)
|
| }
|
| missing = sorted(expected_repos - set(entries_by_repo))
|
| errors.extend(f"missing model evidence: {repo_id}" for repo_id in missing)
|
| manifest_by_repo = {entry.repo_id: entry for entry in expected_entries}
|
| for repo_id in sorted(expected_repos & set(entries_by_repo)):
|
| item = entries_by_repo[repo_id]
|
| entry = manifest_by_repo[repo_id]
|
| if item.get("size") != entry.size:
|
| errors.append(f"{repo_id}: size must be {entry.size}")
|
| if item.get("stage") != entry.stage:
|
| errors.append(f"{repo_id}: stage must be {entry.stage}")
|
| if item.get("base_model") != entry.base_model:
|
| errors.append(f"{repo_id}: base_model must be {entry.base_model}")
|
| if item.get("upstream_license") != entry.upstream_license:
|
| errors.append(f"{repo_id}: upstream_license must be {entry.upstream_license}")
|
| if item.get("artifact_status") not in PUBLIC_ARTIFACT_STATUSES:
|
| errors.append(f"{repo_id}: artifact_status must be public")
|
| if item.get("quality_status") not in PASSING_QUALITY_STATUSES:
|
| errors.append(f"{repo_id}: quality_status must be quality-verified")
|
| if item.get("verifier_passed") is not True:
|
| errors.append(f"{repo_id}: verifier_passed must be true")
|
| limitations = item.get("limitations")
|
| if not isinstance(limitations, list) or not limitations:
|
| errors.append(f"{repo_id}: limitations must be a non-empty list")
|
| if not str(item.get("dataset_repo", "")).strip():
|
| errors.append(f"{repo_id}: dataset_repo is required")
|
| if not str(item.get("training_backend", "")).strip():
|
| errors.append(f"{repo_id}: training_backend is required")
|
| if not str(item.get("source_git_commit", "")).strip():
|
| errors.append(f"{repo_id}: source_git_commit is required")
|
| if not str(item.get("dataset_sha", "")).strip():
|
| errors.append(f"{repo_id}: dataset_sha is required")
|
| if not str(item.get("model_sha", "")).strip():
|
| errors.append(f"{repo_id}: model_sha is required")
|
| for field in ("training_run_url", "model_card_url"):
|
| if not _is_https_url(item.get(field)):
|
| errors.append(f"{repo_id}: {field} must be an HTTPS URL")
|
| for field in ("eval_report_path", "verification_report_path"):
|
| if _resolve_evidence_file(evidence_root, item.get(field)) is None:
|
| errors.append(f"{repo_id}: {field} must point to an evidence file")
|
| metrics = item.get("eval_metrics")
|
| if not isinstance(metrics, dict) or not metrics:
|
| errors.append(f"{repo_id}: eval_metrics must be a non-empty object")
|
| errors.extend(_model_family_dependency_errors(entries_by_repo, manifest_by_repo))
|
| return FullVisionCheck(
|
| name="hf_model_family",
|
| ok=not errors,
|
| detail="Full HF model family exists with verifier-passed quality evidence."
|
| if not errors
|
| else "Full HF model family is incomplete.",
|
| metadata={"repos": repo_results, "evidence": evidence_metadata, "errors": errors},
|
| )
|
|
|
|
|
| def _check_model_card_fields(
|
| *,
|
| repo_id: str,
|
| card: dict[str, Any],
|
| expected_base_model: str,
|
| expected_license: str,
|
| expected_dataset: str,
|
| ) -> list[str]:
|
| """Return errors for required model-card metadata."""
|
| errors: list[str] = []
|
| for field in ("license", "datasets", "base_model"):
|
| if not card.get(field):
|
| errors.append(f"{repo_id}: missing model-card field {field}")
|
| if card.get("base_model") and card.get("base_model") != expected_base_model:
|
| errors.append(f"{repo_id}: base_model must be {expected_base_model}")
|
| if card.get("license") and not license_matches(
|
| card.get("license"),
|
| expected_license,
|
| license_name=card.get("license_name"),
|
| ):
|
| errors.append(f"{repo_id}: license must resolve to {expected_license}")
|
| datasets = card.get("datasets")
|
| if isinstance(datasets, str):
|
| dataset_values = {datasets}
|
| elif isinstance(datasets, list):
|
| dataset_values = {str(item) for item in datasets}
|
| else:
|
| dataset_values = set()
|
| if datasets and expected_dataset not in dataset_values:
|
| errors.append(f"{repo_id}: datasets must include {expected_dataset}")
|
| return errors
|
|
|
|
|
| def _model_family_dependency_errors(
|
| entries_by_repo: dict[str, Any],
|
| manifest_by_repo: dict[str, Any],
|
| ) -> list[str]:
|
| """Return v2 evidence dependency errors across SFT -> GRPO -> GiGPO."""
|
| errors: list[str] = []
|
| for repo_id, entry in manifest_by_repo.items():
|
| predecessor_repo = entry.predecessor_repo
|
| if predecessor_repo is None:
|
| continue
|
| item = entries_by_repo.get(repo_id, {})
|
| predecessor = entries_by_repo.get(predecessor_repo, {})
|
| row_verified = (
|
| item.get("artifact_status") in PUBLIC_ARTIFACT_STATUSES
|
| and item.get("quality_status") in PASSING_QUALITY_STATUSES
|
| )
|
| predecessor_verified = (
|
| predecessor.get("artifact_status") in PUBLIC_ARTIFACT_STATUSES
|
| and predecessor.get("quality_status") in PASSING_QUALITY_STATUSES
|
| )
|
| if row_verified and not predecessor_verified:
|
| errors.append(f"{repo_id}: predecessor {predecessor_repo} must be quality-verified")
|
| return errors
|
|
|
|
|
| def run_full_vision_gate(
|
| *,
|
| evidence_root: Path | None = None,
|
| frontend_url: str = DEFAULT_FRONTEND_URL,
|
| backend_url: str = DEFAULT_BACKEND_URL,
|
| expected_git_sha: str | None = None,
|
| hf_owner: str = DEFAULT_HF_OWNER,
|
| ) -> FullVisionReport:
|
| """Run the external full-original-vision completion gate."""
|
| root = _evidence_root(evidence_root)
|
| checks = [
|
| _check_package_index("pypi", "https://pypi.org/pypi"),
|
| _check_package_index("testpypi", "https://test.pypi.org/pypi"),
|
| _check_publish_evidence(root),
|
| _check_workers_playground(frontend_url, backend_url),
|
| _check_hf_backend(frontend_url, backend_url, expected_git_sha),
|
| _check_dbt_evidence(root),
|
| _check_design_partner_evidence(root),
|
| _check_hf_model_family(root, hf_owner),
|
| ]
|
| return FullVisionReport(ok=all(check.ok for check in checks), checks=checks)
|
|
|
|
|
| def main(argv: list[str] | None = None) -> int:
|
| """Run the full-vision gate as a standalone script."""
|
| import argparse
|
|
|
| parser = argparse.ArgumentParser(description=__doc__)
|
| parser.add_argument("--json", action="store_true", help="Print JSON instead of text.")
|
| parser.add_argument("--evidence-root", type=Path, default=None)
|
| parser.add_argument("--frontend-url", default=DEFAULT_FRONTEND_URL)
|
| parser.add_argument("--backend-url", default=DEFAULT_BACKEND_URL)
|
| parser.add_argument("--expected-git-sha", default=None)
|
| parser.add_argument("--hf-owner", default=DEFAULT_HF_OWNER)
|
| args = parser.parse_args(argv)
|
|
|
| report = run_full_vision_gate(
|
| evidence_root=args.evidence_root,
|
| frontend_url=args.frontend_url,
|
| backend_url=args.backend_url,
|
| expected_git_sha=args.expected_git_sha,
|
| hf_owner=args.hf_owner,
|
| )
|
| if args.json:
|
| print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
|
| else:
|
| for check in report.checks:
|
| status = "ok" if check.ok else "fail"
|
| print(f"{status:4} {check.name}: {check.detail}")
|
| return 0 if report.ok else 1
|
|
|
|
|
| if __name__ == "__main__":
|
| raise SystemExit(main(sys.argv[1:]))
|
|
|
