"""Derive engine-facing feature metadata from the v3 operation registry. ``profile_schema.json.operation_contracts`` is the only feature-operation registry. The engine needs a compact view while validating materialized CDSL, but that view is intentionally derived at load time instead of maintained as a second hand-written ``feature_atomic_ids`` table. """ from __future__ import annotations from typing import Any def materialized_feature_contracts(profile: dict[str, Any]) -> dict[str, dict[str, Any]]: """Return required engine metadata derived from each v3 operation contract. Author-owned params are supplied by the tool schema. Server-injected ``params.*`` paths become required only after materialization, while injected feature selectors are represented outside the params object. """ raw_contracts = profile.get("operation_contracts") if not isinstance(raw_contracts, dict) or not raw_contracts: raise ValueError("profile schema has no operation_contracts registry") derived: dict[str, dict[str, Any]] = {} for atomic_id, raw in raw_contracts.items(): if not isinstance(atomic_id, str) or not atomic_id or not isinstance(raw, dict): raise ValueError("operation contract registry has an invalid entry") if raw.get("atomic_id") != atomic_id: raise ValueError(f"operation contract key and atomic_id disagree for {atomic_id}") shape = raw.get("fragment_shape") params_schema = raw.get("author_params_schema") injected_paths = raw.get("server_injected_paths") selector_policy = raw.get("selector_policy") if ( not isinstance(shape, dict) or not isinstance(params_schema, dict) or not isinstance(injected_paths, list) or not isinstance(selector_policy, dict) ): raise ValueError(f"operation contract is incomplete for {atomic_id}") properties = params_schema.get("properties") required = params_schema.get("required") if not isinstance(properties, dict) or not isinstance(required, list): raise ValueError(f"operation author params are invalid for {atomic_id}") if not all(isinstance(name, str) and name in properties for name in required): raise ValueError(f"operation required params are invalid for {atomic_id}") materialized_required = list(dict.fromkeys(required)) for path in injected_paths: if not isinstance(path, str): raise ValueError(f"operation injection path is invalid for {atomic_id}") if path.startswith("params.") and path.count(".") == 1: materialized_required.append(path.removeprefix("params.")) elif path != "feature.selectors": raise ValueError(f"operation injection path is unsupported for {atomic_id}") materialized_required = list(dict.fromkeys(materialized_required)) derived[atomic_id] = { "required_params": materialized_required, "optional_params": [name for name in properties if name not in materialized_required], "requires_sketch": shape.get("sketch") == "required", "selector_slot": selector_policy.get("slot"), "selector_token_kind": selector_policy.get("token_kind"), } return derived