"""Canonical config loader with schema validation. Every config file is loaded via `validate_and_load(path, schema_name)`. Unknown / typo'd / wrong-typed / missing-required fields raise SchemaValidationError at load time, NOT at first use. Per Tenet 6 (architectural-law.md:186-263) — fail-loud is the default. Adding a new config file: 1. Define a Pydantic model in this file. 2. Register it in SCHEMAS. 3. Call validate_and_load(path, schema_name) from the loader. NOTE: this file deliberately does NOT use `from __future__ import annotations`. Pydantic v2 needs annotations resolvable at class-definition time to build its validators, and when the module is loaded via importlib proxy (pipeline/lib/config_schema.py), the deferred-string-annotation form breaks `model_validate` with "class is not fully defined" errors. """ import json from pathlib import Path from typing import Any, Optional, Type import yaml from pydantic import BaseModel, ConfigDict, ValidationError from .exceptions import SchemaValidationError class _BaseConfigSchema(BaseModel): """Base for every Recoil config schema. Default: extra='forbid' so typos at the top level fail loud. Subclasses with extension fields (model_profiles entries, etc.) override. """ model_config = ConfigDict( extra="forbid", frozen=False, validate_assignment=True, ) # ────────────────────────────────────────────────────────────────────── # 1. pipeline_config.json # ────────────────────────────────────────────────────────────────────── class PipelineConfigSchema(_BaseConfigSchema): """Schema for config/pipeline_config.json (170-line top-level config).""" schema_version: int # Path/root settings projects_root: Optional[str] = None visual_state_namespace: Optional[str] = None recoil_engine_root: Optional[str] = None output_root: Optional[str] = None # Default model selection default_model: Optional[str] = None exploration_model: Optional[str] = None fallback_exploration_model: Optional[str] = None # Image / aspect settings default_image_size: Optional[str] = None production_aspect_ratio: Optional[str] = None grid_aspect_ratio: Optional[str] = None storyboard_iteration: Optional[dict[str, Any]] = None max_references_per_shot: Optional[int] = None max_concurrent_api_calls: Optional[int] = None # Nested sections — kept as dicts; values vary widely per section visual_defaults: Optional[dict[str, Any]] = None complexity_tiers: Optional[dict[str, Any]] = None video: Optional[dict[str, Any]] = None routing: Optional[dict[str, Any]] = None model_capabilities: Optional[dict[str, Any]] = None coverage_strategy: Optional[dict[str, Any]] = None critic_flags: Optional[dict[str, bool]] = None # Boolean / scalar feature flags + tunables enable_scene_visual_locks: Optional[bool] = None enable_moodboard_to_text: Optional[bool] = None location_ref_mode: Optional[str] = None previz_temperature: Optional[float] = None # Top-level remains permissive so future additions don't break loaders model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 2. model_profiles.json # ────────────────────────────────────────────────────────────────────── class ModelProfilesSchema(_BaseConfigSchema): """Schema for config/model_profiles.json. Top-level keys are model_ids → entry dicts. Entries vary widely between image and video models, so values are kept as `dict[str, Any]` and only the `schema_version` field is typed at the root. """ schema_version: int model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 3. provider_strategy.json # ────────────────────────────────────────────────────────────────────── class ProviderStrategySchema(_BaseConfigSchema): """Schema for config/provider_strategy.json. Top-level: schema_version + one entry per model_id with shape: { "capability_exceptions": dict, "primary": str, "primary_tier": str, } """ schema_version: int model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 4. pricing_rates.json # ────────────────────────────────────────────────────────────────────── class PricingRatesSchema(_BaseConfigSchema): """Schema for config/pricing_rates.json. Top-level: schema_version, note, and rate_cards (list of date-stamped rate cards). """ schema_version: int note: Optional[str] = None rate_cards: Optional[list[dict[str, Any]]] = None model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 5. model_roles.json # ────────────────────────────────────────────────────────────────────── class ModelRolesSchema(_BaseConfigSchema): """Schema for config/model_roles.json. Top-level: role-buckets (image/video/text/tts/qc), each a mapping of role_name → model_id. Plus a `_doc` documentation string preserved via extra='allow' (Pydantic v2 ignores _-prefixed field declarations). """ schema_version: int image: Optional[dict[str, str]] = None video: Optional[dict[str, str]] = None text: Optional[dict[str, str]] = None tts: Optional[dict[str, str]] = None qc: Optional[dict[str, str]] = None model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 6. prompt_constants.json # ────────────────────────────────────────────────────────────────────── class PromptConstantsSchema(_BaseConfigSchema): """Schema for config/prompt_constants.json. Top-level sections: production, casting, shared, formatter_limits. Each section is a flat string→string (or string→limits-dict for formatter_limits) map. Plus a `_doc` documentation string preserved via extra='allow'. """ schema_version: int production: Optional[dict[str, str]] = None casting: Optional[dict[str, str]] = None shared: Optional[dict[str, str]] = None formatter_limits: Optional[dict[str, dict[str, Any]]] = None model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 7. lexicon.json # ────────────────────────────────────────────────────────────────────── class LexiconSchema(_BaseConfigSchema): """Schema for config/lexicon.json. Top-level: kinetic_map (list of {id, pattern, descriptor}), fallback (str), lighting_direction_map and lighting_quality_map (lists of {pattern, direction|quality}). Plus a `_doc` documentation string preserved via extra='allow'. """ schema_version: int kinetic_map: Optional[list[dict[str, Any]]] = None fallback: Optional[str] = None lighting_direction_map: Optional[list[dict[str, Any]]] = None lighting_quality_map: Optional[list[dict[str, Any]]] = None model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 8. ref_descriptors.json # ────────────────────────────────────────────────────────────────────── class RefDescriptorsSchema(_BaseConfigSchema): """Schema for config/ref_descriptors.json. Top-level keys are ref-class names (character, location, wardrobe, hair_makeup, props), each mapping to a per-class descriptor block. Descriptor blocks vary by class (composite_grid vs parallel_singles strategies) so values stay as `dict[str, Any]`. """ schema_version: int character: Optional[dict[str, Any]] = None location: Optional[dict[str, Any]] = None wardrobe: Optional[dict[str, Any]] = None hair_makeup: Optional[dict[str, Any]] = None props: Optional[dict[str, Any]] = None model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # 9. PROMPT_BIBLE.yaml # ────────────────────────────────────────────────────────────────────── class PromptBibleSchema(_BaseConfigSchema): """Schema for config/PROMPT_BIBLE.yaml. Top-level keys are model_ids (seedream-v4.5, kling-v3, veo-3.1, ...) → per-model rule dicts (meta, prompt, refs, aspect_ratio, first_last_frame, duration, audio, gotchas). Per-model dicts vary widely so values stay as `dict[str, Any]`. """ schema_version: int model_config = ConfigDict(extra="allow") # ────────────────────────────────────────────────────────────────────── # Registry # ────────────────────────────────────────────────────────────────────── SCHEMAS: dict[str, Type[_BaseConfigSchema]] = { "pipeline_config": PipelineConfigSchema, "model_profiles": ModelProfilesSchema, "provider_strategy": ProviderStrategySchema, "pricing_rates": PricingRatesSchema, "model_roles": ModelRolesSchema, "prompt_constants": PromptConstantsSchema, "lexicon": LexiconSchema, "ref_descriptors": RefDescriptorsSchema, "prompt_bible": PromptBibleSchema, } def validate_and_load(path: Path, schema_name: str) -> dict[str, Any]: """Load a config file and validate against the registered schema. Args: path: Path to the config file (.json or .yaml). schema_name: Key in SCHEMAS dict. Returns: Validated config as a dict (model.model_dump()). Raises: SchemaValidationError: on shape mismatch, missing required field, wrong type, unknown extra field (where extra='forbid'), unknown schema_name, or parse failure. FileNotFoundError: if path does not exist. """ if schema_name not in SCHEMAS: raise SchemaValidationError( f"Unknown schema_name {schema_name!r}. " f"Registered: {sorted(SCHEMAS.keys())}" ) if not path.is_file(): raise FileNotFoundError(f"Config file not found: {path}") try: if path.suffix in (".yaml", ".yml"): raw = yaml.safe_load(path.read_text(encoding="utf-8")) else: raw = json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, yaml.YAMLError) as e: raise SchemaValidationError( f"Config file {path} failed to parse: {e}" ) from e schema_cls = SCHEMAS[schema_name] try: model = schema_cls.model_validate(raw) except ValidationError as e: raise SchemaValidationError( f"Config file {path} (schema {schema_name!r}) failed validation:\n" f"{e}" ) from e # exclude_none=True drops keys whose value is None at every level — # preserves the legacy "only keys actually in the file" dict shape # for callers that don't expect Pydantic-default None synthesizations. return model.model_dump(exclude_none=True) __all__ = [ "SCHEMAS", "SchemaValidationError", "validate_and_load", "PipelineConfigSchema", "ModelProfilesSchema", "ProviderStrategySchema", "PricingRatesSchema", "ModelRolesSchema", "PromptConstantsSchema", "LexiconSchema", "RefDescriptorsSchema", "PromptBibleSchema", ]