#!/usr/bin/env python3 """Validate Vercy control files, runtime records and semantic invariants.""" from __future__ import annotations import argparse import hashlib import json import re from pathlib import Path from vercy_runtime import ( VercyError, load_json, load_records, load_schema, object_heads, parse_time, resolve_facts, validate_schema, ) REQUIRED = [ "AGENTS.md", "dimension.yaml", "registries/meta-models.yaml", "registries/meta-objects.yaml", "registries/events.yaml", "policies/access.yaml", "policies/lifecycle.md", "policies/autonomy.yaml", "policies/model-deployment.yaml", "policies/conflict-resolution.yaml", "bindings/storage.yaml", "memory.yaml", "federation.yaml", "context-routing.yaml", "compatibility.yaml", "migrations/index.yaml", "registries/model-links.yaml", "registries/model-requests.yaml", "vercy.lock", ] JSON_CONTROL = [rel for rel in REQUIRED if Path(rel).suffix in {".yaml", ".lock"}] PLACEHOLDER = re.compile(r"\{\{[A-Z0-9_]+\}\}") def issue(rule: str, location: str, message: str) -> dict[str, str]: return {"rule": rule, "location": location, "message": message} def report(root: Path, errors: list[dict[str, str]], warnings: list[dict[str, str]], counts: dict[str, int]) -> dict[str, object]: return { "valid": not errors, "dimension": str(root), "conformanceLevel": "V3" if not errors else "failed", "schemas": "https://ver.cy/schemas/dimension/1.0/", "counts": counts, "errors": errors, "warnings": warnings, "scope": "executable structure and record semantics; does not certify external truth, legal authority, or safety", } def validate(root: Path) -> dict[str, object]: errors: list[dict[str, str]] = [] warnings: list[dict[str, str]] = [] counts = {"objects": 0, "facts": 0, "relations": 0, "events": 0} if not root.is_dir(): errors.append(issue("V0-DIR", ".", f"not a directory: {root}")) return report(root, errors, warnings, counts) for rel in REQUIRED: path = root / rel if not path.is_file(): errors.append(issue("V0-REQUIRED", rel, "required file is missing")) continue hits = sorted(set(PLACEHOLDER.findall(path.read_text(encoding="utf-8")))) if hits: errors.append(issue("V0-PLACEHOLDER", rel, f"unresolved placeholders: {', '.join(hits)}")) controls: dict[str, object] = {} for rel in JSON_CONTROL: path = root / rel if not path.is_file(): continue try: controls[rel] = load_json(path) except VercyError as exc: errors.append(issue("V0-CONTROL-PROFILE", rel, str(exc))) agents = root / "AGENTS.md" if agents.is_file() and "Vercy Dimension" not in agents.read_text(encoding="utf-8"): errors.append(issue("V0-AGENTS", "AGENTS.md", "file does not identify a Vercy Dimension")) dimension = controls.get("dimension.yaml") if isinstance(dimension, dict): for message in validate_schema(dimension, load_schema("dimension")): errors.append(issue("V1-DIMENSION-SCHEMA", "dimension.yaml", message)) dimension_id = dimension.get("id") for rel, value in controls.items(): if isinstance(value, dict) and "dimension" in value and value["dimension"] != dimension_id: errors.append(issue("V2-DIMENSION-IDENTITY", rel, "dimension id does not match manifest")) for rel, schema_name in ( ("policies/conflict-resolution.yaml", "conflict-policy"), ("compatibility.yaml", "compatibility"), ): value = controls.get(rel) if isinstance(value, dict): for message in validate_schema(value, load_schema(schema_name)): errors.append(issue("V1-CONTROL-SCHEMA", rel, message)) conflict_policy = controls.get("policies/conflict-resolution.yaml") if isinstance(conflict_policy, dict) and conflict_policy.get("precedence") != ["explicit-supersedes", "higher-authority-rank", "same-authority-policy"]: errors.append(issue("V3-CONFLICT-PRECEDENCE", "policies/conflict-resolution.yaml", "precedence must be explicit-supersedes, higher-authority-rank, same-authority-policy")) lock = controls.get("vercy.lock") if isinstance(lock, dict) and lock.get("preset", {}).get("id") == "vercy.preset.ai-subject": errors.append(issue("V3-AI-SUBJECT-GATE", "vercy.lock", "reserved AI-subject preset is not instantiable")) records: list[tuple[str, Path, dict[str, object]]] = [] try: records = load_records(root) except VercyError as exc: errors.append(issue("V1-RECORD-READ", "data", str(exc))) ids: dict[str, str] = {} objects: list[dict[str, object]] = [] facts: list[dict[str, object]] = [] relations: list[dict[str, object]] = [] events: list[dict[str, object]] = [] id_fields = {"object": "recordId", "fact": "factId", "relation": "relationId", "event": "eventId"} buckets = {"object": objects, "fact": facts, "relation": relations, "event": events} for kind, path, value in records: rel = path.relative_to(root).as_posix() for message in validate_schema(value, load_schema(kind)): errors.append(issue("V1-RECORD-SCHEMA", rel, message)) record_id = value.get(id_fields[kind]) if isinstance(record_id, str): if record_id in ids: errors.append(issue("V2-ID-UNIQUE", rel, f"id duplicates {ids[record_id]}")) ids[record_id] = rel buckets[kind].append(value) counts["objects" if kind == "object" else f"{kind}s"] += 1 object_ids = {str(item.get("objectId")) for item in objects} object_record_ids = {str(item.get("recordId")) for item in objects} fact_ids = {str(item.get("factId")) for item in facts} path_contracts: dict[str, dict[str, object]] = {} models = controls.get("registries/meta-models.yaml") if isinstance(models, dict): for index, model in enumerate(models.get("models", [])): location = f"registries/meta-models.yaml.models[{index}]" if not isinstance(model, dict): errors.append(issue("V1-MODEL-ENTRY", location, "model entry must be an object")) continue for field in ("id", "version", "agents", "specification", "runtimeSchema", "specificationDigest"): if not isinstance(model.get(field), str) or not model.get(field): errors.append(issue("V1-MODEL-ENTRY", location, f"required string {field!r} is missing")) for field in ("agents", "specification", "runtimeSchema"): reference = model.get(field) if not isinstance(reference, str) or reference.startswith(("http://", "https://")): errors.append(issue("V2-MODEL-RUNTIME", location, f"reference {field!r} must be a local pinned file for executable validation")) continue resolved = (root / reference).resolve() if root not in resolved.parents or not resolved.is_file(): errors.append(issue("V2-MODEL-RUNTIME", location, f"reference {field!r} does not resolve inside the Dimension")) spec_ref = model.get("specification") expected_digest = model.get("specificationDigest") if isinstance(spec_ref, str) and isinstance(expected_digest, str) and not spec_ref.startswith(("http://", "https://")): spec_path = (root / spec_ref).resolve() if spec_path.is_file(): actual = "sha256:" + hashlib.sha256(spec_path.read_bytes()).hexdigest() if actual != expected_digest: errors.append(issue("V2-MODEL-DIGEST", location, f"specification digest mismatch: expected {expected_digest}, got {actual}")) runtime_ref = model.get("runtimeSchema") if isinstance(runtime_ref, str) and not runtime_ref.startswith(("http://", "https://")): runtime_path = (root / runtime_ref).resolve() if runtime_path.is_file(): try: runtime_schema = load_json(runtime_path) for message in validate_schema(runtime_schema, load_schema("runtime-model")): errors.append(issue("V1-MODEL-RUNTIME-SCHEMA", runtime_ref, message)) if runtime_schema.get("modelId") != model.get("id"): errors.append(issue("V2-MODEL-RUNTIME-ID", runtime_ref, "modelId does not match registry entry")) for fact_path, contract in runtime_schema.get("paths", {}).items(): if fact_path in path_contracts and path_contracts[fact_path] != contract: errors.append(issue("V2-MODEL-PATH-COLLISION", runtime_ref, f"path {fact_path!r} has incompatible contracts")) path_contracts[fact_path] = contract except VercyError as exc: errors.append(issue("V1-MODEL-RUNTIME-SCHEMA", runtime_ref, str(exc))) for item in objects: parent = item.get("previousRecordId") if parent and parent not in object_record_ids: errors.append(issue("V2-OBJECT-CHAIN", ids.get(str(item.get("recordId")), "data/objects"), f"previousRecordId {parent!r} does not resolve")) for object_id, heads in object_heads(objects).items(): if len(heads) > 1: errors.append(issue("V2-PARALLEL-WRITE", "data/objects", f"object {object_id!r} has competing heads {heads}; create an explicit merge revision")) for item in facts: rel = ids.get(str(item.get("factId")), "data/facts") if item.get("subjectId") not in object_ids: errors.append(issue("V2-FACT-SUBJECT", rel, f"subjectId {item.get('subjectId')!r} does not resolve")) for old in item.get("supersedes", []): if old not in fact_ids: errors.append(issue("V2-FACT-SUPERSEDES", rel, f"superseded fact {old!r} does not resolve")) fact_path = item.get("path") contract = path_contracts.get(str(fact_path)) if contract is None: errors.append(issue("V2-FACT-PATH", rel, f"path {fact_path!r} is not declared by an installed model runtime schema")) else: value = item.get("value") value_type = "boolean" if isinstance(value, bool) else "integer" if isinstance(value, int) else "number" if isinstance(value, float) else "string" if isinstance(value, str) else "array" if isinstance(value, list) else "object" if isinstance(value, dict) else "null" allowed_types = contract.get("valueTypes", []) if value_type not in allowed_types and not (value_type == "integer" and "number" in allowed_types): errors.append(issue("V2-FACT-VALUE", rel, f"path {fact_path!r} does not allow value type {value_type!r}")) if item.get("unit") not in contract.get("units", [None]): errors.append(issue("V2-FACT-UNIT", rel, f"path {fact_path!r} does not allow unit {item.get('unit')!r}")) if item.get("validTo"): try: if parse_time(str(item["validTo"])) <= parse_time(str(item["validFrom"])): errors.append(issue("V2-TEMPORAL-INTERVAL", rel, "validTo must be later than validFrom")) except VercyError: pass for item in relations: rel = ids.get(str(item.get("relationId")), "data/relations") for field in ("sourceId", "targetId"): if item.get(field) not in object_ids: errors.append(issue("V2-RELATION-ENDPOINT", rel, f"{field} {item.get(field)!r} does not resolve")) for item in events: rel = ids.get(str(item.get("eventId")), "data/events") for subject in item.get("subjectIds", []): if subject not in object_ids and subject not in ids: errors.append(issue("V2-EVENT-SUBJECT", rel, f"subject {subject!r} does not resolve")) policy = controls.get("policies/conflict-resolution.yaml") if isinstance(policy, dict): try: resolutions = resolve_facts(facts, policy) for item in resolutions: if item["resolution"] == "contested": warnings.append(issue("V3-CURRENT-TRUTH-CONTESTED", f"{item['subjectId']}:{item['path']}", "multiple current assertions remain visible; no value was guessed")) except VercyError as exc: errors.append(issue("V3-CURRENT-TRUTH", "data/facts", str(exc))) if isinstance(models, dict) and not models.get("models"): warnings.append(issue("V2-MODEL-CLOSURE", "registries/meta-models.yaml", "no meta-model is installed yet")) return report(root, errors, warnings, counts) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("dimension", help="Dimension directory") args = parser.parse_args() root = Path(args.dimension).expanduser().resolve() result = validate(root) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["valid"] else 1 if __name__ == "__main__": raise SystemExit(main())