"""Load the JSON Schema contracts and validate canon logs / origin seeds. Schemas are the single source of truth in /docs/schemas. This module is the api's half of the contract (charter ยง11): every canon log the client posts is validated here before it could ever reach a prompt. """ import json import os from functools import lru_cache from pathlib import Path from jsonschema import Draft202012Validator def _schema_dir() -> Path: """Locate the schema directory. Honours CANON_SCHEMA_DIR, else walks up from this file looking for docs/schemas (local checkout) or schemas (bundled into the Docker image). """ env = os.environ.get("CANON_SCHEMA_DIR") if env: return Path(env) here = Path(__file__).resolve() for parent in here.parents: for candidate in (parent / "docs" / "schemas", parent / "schemas"): if candidate.is_dir(): return candidate raise RuntimeError("schema directory not found") @lru_cache(maxsize=None) def _validator(schema_name: str) -> Draft202012Validator: with open(_schema_dir() / schema_name) as f: return Draft202012Validator(json.load(f)) def validation_errors(doc: dict, schema_name: str) -> list[str]: validator = _validator(schema_name) # Stringify path segments before sorting: a path can mix str property names # and int array indices, which are not orderable against each other. return [e.message for e in sorted(validator.iter_errors(doc), key=lambda e: [str(p) for p in e.path])] def validate_origin(doc: dict) -> list[str]: return validation_errors(doc, "origin.schema.json") def validate_canon_log(doc: dict) -> list[str]: return validation_errors(doc, "canon-log.schema.json")