176 lines
7.3 KiB
Python
176 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate a generated fal model registry before it replaces the baseline.
|
|
|
|
The scheduled registry refresh uses this dependency-free gate to reject
|
|
truncated catalog responses, duplicate or malformed records, nondeterministic
|
|
ordering, and unexpectedly large endpoint changes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REQUIRED_TOP_LEVEL = {
|
|
"version",
|
|
"models",
|
|
"model_count",
|
|
"live_model_count",
|
|
"deprecated_model_count",
|
|
}
|
|
REQUIRED_MODEL_FIELDS = {
|
|
"endpoint_id",
|
|
"title",
|
|
"category",
|
|
"description",
|
|
"family",
|
|
"lab",
|
|
"pricing",
|
|
"published_at",
|
|
"inputs",
|
|
"output_kind",
|
|
"output_props",
|
|
"thumbnail",
|
|
}
|
|
OUTPUT_KINDS = {"audio", "file", "image", "images", "json", "text", "video"}
|
|
ENDPOINT_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
|
|
|
|
|
|
class RegistryValidationError(ValueError):
|
|
"""Raised when a registry cannot safely be promoted."""
|
|
|
|
|
|
def load_registry(path: Path) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise RegistryValidationError(f"Could not read {path}: {exc}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise RegistryValidationError(f"{path} must contain a JSON object")
|
|
return payload
|
|
|
|
|
|
def validate_registry(registry: dict[str, Any], *, min_models: int = 500) -> set[str]:
|
|
missing_top = REQUIRED_TOP_LEVEL - registry.keys()
|
|
if missing_top:
|
|
raise RegistryValidationError(f"Registry is missing top-level fields: {sorted(missing_top)}")
|
|
|
|
if registry["version"] != 1:
|
|
raise RegistryValidationError(f"Unsupported registry version: {registry['version']!r}")
|
|
|
|
models = registry["models"]
|
|
if not isinstance(models, list):
|
|
raise RegistryValidationError("Registry 'models' must be a list")
|
|
if registry["model_count"] != len(models):
|
|
raise RegistryValidationError(f"model_count={registry['model_count']} does not match {len(models)} records")
|
|
deprecated_count = sum(bool(model.get("deprecated")) for model in models if isinstance(model, dict))
|
|
live_count = len(models) - deprecated_count
|
|
if registry["live_model_count"] != live_count:
|
|
raise RegistryValidationError(
|
|
f"live_model_count={registry['live_model_count']} does not match {live_count} records"
|
|
)
|
|
if registry["deprecated_model_count"] != deprecated_count:
|
|
raise RegistryValidationError(
|
|
f"deprecated_model_count={registry['deprecated_model_count']} does not match "
|
|
f"{deprecated_count} records"
|
|
)
|
|
if len(models) < min_models:
|
|
raise RegistryValidationError(f"Registry has only {len(models)} models; minimum is {min_models}")
|
|
|
|
endpoint_ids: list[str] = []
|
|
for index, model in enumerate(models):
|
|
if not isinstance(model, dict):
|
|
raise RegistryValidationError(f"Model {index} is not an object")
|
|
missing = REQUIRED_MODEL_FIELDS - model.keys()
|
|
if missing:
|
|
raise RegistryValidationError(f"Model {index} is missing fields: {sorted(missing)}")
|
|
endpoint_id = model["endpoint_id"]
|
|
if (
|
|
not isinstance(endpoint_id, str)
|
|
or not endpoint_id.strip()
|
|
or not ENDPOINT_ID_PATTERN.fullmatch(endpoint_id)
|
|
):
|
|
raise RegistryValidationError(f"Model {index} has an invalid endpoint_id")
|
|
if not isinstance(model["title"], str) or not model["title"].strip():
|
|
raise RegistryValidationError(f"{endpoint_id} has an invalid title")
|
|
if not isinstance(model["inputs"], list):
|
|
raise RegistryValidationError(f"{endpoint_id} inputs must be a list")
|
|
if not isinstance(model["output_props"], list):
|
|
raise RegistryValidationError(f"{endpoint_id} output_props must be a list")
|
|
if model["output_kind"] not in OUTPUT_KINDS:
|
|
raise RegistryValidationError(f"{endpoint_id} has unknown output_kind={model['output_kind']!r}")
|
|
if "deprecated" in model and not isinstance(model["deprecated"], bool):
|
|
raise RegistryValidationError(f"{endpoint_id} deprecated must be a boolean")
|
|
endpoint_ids.append(endpoint_id)
|
|
|
|
if len(endpoint_ids) != len(set(endpoint_ids)):
|
|
duplicates = sorted(endpoint_id for endpoint_id in set(endpoint_ids) if endpoint_ids.count(endpoint_id) > 1)
|
|
raise RegistryValidationError(f"Duplicate endpoint IDs: {duplicates[:10]}")
|
|
if endpoint_ids != sorted(endpoint_ids):
|
|
raise RegistryValidationError("Models must be sorted by endpoint_id")
|
|
return set(endpoint_ids)
|
|
|
|
|
|
def compare_registries(
|
|
baseline_ids: set[str],
|
|
candidate_ids: set[str],
|
|
*,
|
|
max_removal_fraction: float = 0.05,
|
|
max_addition_fraction: float = 0.25,
|
|
allow_large_change: bool = False,
|
|
) -> tuple[set[str], set[str]]:
|
|
if not 0 <= max_removal_fraction <= 1:
|
|
raise RegistryValidationError("max_removal_fraction must be between 0 and 1")
|
|
if not 0 <= max_addition_fraction <= 1:
|
|
raise RegistryValidationError("max_addition_fraction must be between 0 and 1")
|
|
added = candidate_ids - baseline_ids
|
|
removed = baseline_ids - candidate_ids
|
|
removal_fraction = len(removed) / len(baseline_ids) if baseline_ids else 0.0
|
|
addition_fraction = len(added) / len(baseline_ids) if baseline_ids else 0.0
|
|
if removal_fraction > max_removal_fraction and not allow_large_change:
|
|
raise RegistryValidationError(
|
|
f"Candidate removes {len(removed)}/{len(baseline_ids)} endpoints "
|
|
f"({removal_fraction:.1%}), above the {max_removal_fraction:.1%} limit"
|
|
)
|
|
if addition_fraction > max_addition_fraction and not allow_large_change:
|
|
raise RegistryValidationError(
|
|
f"Candidate adds {len(added)}/{len(baseline_ids)} endpoints "
|
|
f"({addition_fraction:.1%}), above the {max_addition_fraction:.1%} limit"
|
|
)
|
|
return added, removed
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("candidate", type=Path, help="Generated registry to validate")
|
|
parser.add_argument("--baseline", type=Path, help="Committed registry to compare")
|
|
parser.add_argument("--min-models", type=int, default=500)
|
|
parser.add_argument("--max-removal-fraction", type=float, default=0.05)
|
|
parser.add_argument("--max-addition-fraction", type=float, default=0.25)
|
|
parser.add_argument("--allow-large-change", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
candidate_ids = validate_registry(load_registry(args.candidate), min_models=args.min_models)
|
|
added: set[str] = set()
|
|
removed: set[str] = set()
|
|
if args.baseline:
|
|
baseline_ids = validate_registry(load_registry(args.baseline), min_models=args.min_models)
|
|
added, removed = compare_registries(
|
|
baseline_ids,
|
|
candidate_ids,
|
|
max_removal_fraction=args.max_removal_fraction,
|
|
max_addition_fraction=args.max_addition_fraction,
|
|
allow_large_change=args.allow_large_change,
|
|
)
|
|
print(f"Registry valid: {len(candidate_ids)} models " f"(+{len(added)} / -{len(removed)} vs baseline)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|