Gökay Aydoğan
647a38f265
* fix: preserve and validate model controls across the fal catalog * fix: preserve enum constraints through nested normalization
239 lines
11 KiB
Python
239 lines
11 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"}
|
|
INPUT_TYPES = {"string", "integer", "number", "boolean", "enum", "object", "array", "json"}
|
|
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")
|
|
input_names = set()
|
|
for inp in model["inputs"]:
|
|
if not isinstance(inp, dict) or not isinstance(inp.get("name"), str) or not inp["name"]:
|
|
raise RegistryValidationError(f"{endpoint_id} has an invalid input record")
|
|
name = inp["name"]
|
|
if name in input_names:
|
|
raise RegistryValidationError(f"{endpoint_id} has duplicate input {name}")
|
|
input_names.add(name)
|
|
if inp.get("type") not in INPUT_TYPES:
|
|
raise RegistryValidationError(f"{endpoint_id}.{name} has an invalid input type")
|
|
if not isinstance(inp.get("required"), bool):
|
|
raise RegistryValidationError(f"{endpoint_id}.{name} required must be a boolean")
|
|
if inp["type"] == "enum" and (not isinstance(inp.get("enum"), list) or not inp["enum"]):
|
|
raise RegistryValidationError(f"{endpoint_id}.{name} has no enum choices")
|
|
if "suggestions" in inp and (
|
|
inp["type"] != "string" or not isinstance(inp["suggestions"], list)
|
|
or not inp["suggestions"] or not all(isinstance(v, str) for v in inp["suggestions"])
|
|
):
|
|
raise RegistryValidationError(f"{endpoint_id}.{name} has invalid suggestions")
|
|
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 compare_model_inputs(baseline: dict[str, Any], candidate: dict[str, Any]) -> list[str]:
|
|
"""Find controls or choices that a refresh would remove from existing nodes.
|
|
|
|
Intentional upstream removals require review; silently accepting a partial
|
|
schema can otherwise publish missing duration/resolution controls as valid.
|
|
"""
|
|
previous = {model["endpoint_id"]: model for model in baseline["models"]}
|
|
regressions = []
|
|
for model in candidate["models"]:
|
|
endpoint_id = model["endpoint_id"]
|
|
if endpoint_id not in previous:
|
|
continue
|
|
current = {inp["name"]: inp for inp in model["inputs"]}
|
|
for old in previous[endpoint_id]["inputs"]:
|
|
name = old["name"]
|
|
new = current.get(name)
|
|
if new is None:
|
|
regressions.append(f"{endpoint_id}: removed input {name}")
|
|
elif old["type"] in ("integer", "number", "boolean") and new["type"] == "json":
|
|
regressions.append(f"{endpoint_id}.{name}: lost {old['type']} control")
|
|
elif old["type"] == "enum" or old.get("suggestions"):
|
|
choices = new.get("enum") if new["type"] == "enum" else new.get("suggestions")
|
|
if not choices:
|
|
control = "enum control" if old["type"] == "enum" else "suggested choices"
|
|
regressions.append(f"{endpoint_id}.{name}: lost {control}")
|
|
else:
|
|
old_choices = old["enum"] if old["type"] == "enum" else old["suggestions"]
|
|
lost = [value for value in old_choices if value not in choices]
|
|
if lost:
|
|
regressions.append(f"{endpoint_id}.{name}: removed choices {lost}")
|
|
return regressions
|
|
|
|
|
|
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")
|
|
parser.add_argument("--allow-input-removal", action="store_true",
|
|
help="Allow reviewed removals of existing input controls or enum choices")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
candidate = load_registry(args.candidate)
|
|
candidate_ids = validate_registry(candidate, min_models=args.min_models)
|
|
added: set[str] = set()
|
|
removed: set[str] = set()
|
|
if args.baseline:
|
|
baseline = load_registry(args.baseline)
|
|
baseline_ids = validate_registry(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,
|
|
)
|
|
regressions = compare_model_inputs(baseline, candidate)
|
|
if regressions and not args.allow_input_removal:
|
|
raise RegistryValidationError(
|
|
"Candidate removes existing controls; review before using --allow-input-removal:\n"
|
|
+ "\n".join(regressions[:20])
|
|
)
|
|
print(f"Registry valid: {len(candidate_ids)} models " f"(+{len(added)} / -{len(removed)} vs baseline)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|