labels.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """``model_tiers`` registry loader and validation (§PR.3, §PC.3).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | from functools import lru_cache |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import yaml |
| 10 | |
| 11 | from adapters.errors import ConfigError |
| 12 | from tools.freeze_reviewer.labels import is_vendor_slug |
| 13 | |
| 14 | KEBAB_ID_RE = re.compile(r"^[a-z][a-z0-9]*(-[a-z0-9]+)*$") |
| 15 | MODEL_TIER_ENTRY_KEYS = frozenset({"id", "display", "meaning", "cursor_model_hint", "cost_class"}) |
| 16 | HUMAN_TIER = "human" |
| 17 | COST_CLASS_VALUES = frozenset({"free", "low", "moderate", "high"}) |
| 18 | |
| 19 | |
| 20 | class RoutingPolicyError(Exception): |
| 21 | """Routing policy load/validation failure with frozen exit codes.""" |
| 22 | |
| 23 | def __init__(self, message: str, *, exit_code: int = 30, citation: str | None = None) -> None: |
| 24 | self.message = message |
| 25 | self.exit_code = exit_code |
| 26 | self.citation = citation |
| 27 | super().__init__(message) |
| 28 | |
| 29 | |
| 30 | def _validate_cost_class_value( |
| 31 | value: object, |
| 32 | *, |
| 33 | prefix: str, |
| 34 | path: str, |
| 35 | fail_closed: bool, |
| 36 | ) -> None: |
| 37 | if value is None: |
| 38 | return |
| 39 | if not isinstance(value, str): |
| 40 | message = f"{prefix}.cost_class must be a string" |
| 41 | if fail_closed: |
| 42 | raise RoutingPolicyError(message, exit_code=32, citation=path) |
| 43 | raise ConfigError(message, path) |
| 44 | if value not in COST_CLASS_VALUES: |
| 45 | message = ( |
| 46 | f"{prefix}.cost_class {value!r} outside frozen vocabulary " |
| 47 | f"{sorted(COST_CLASS_VALUES)}" |
| 48 | ) |
| 49 | if fail_closed: |
| 50 | raise RoutingPolicyError(message, exit_code=32, citation=path) |
| 51 | raise ConfigError(message, path) |
| 52 | |
| 53 | |
| 54 | @lru_cache(maxsize=4) |
| 55 | def load_model_tier_ids(kit_root: Path) -> frozenset[str]: |
| 56 | """Load allowed ``model_tiers[].id`` values from kit-carried policy.""" |
| 57 | path = kit_root / "policy" / "model-labels.yaml" |
| 58 | if not path.is_file(): |
| 59 | raise ConfigError("model tier registry missing", str(path)) |
| 60 | raw = yaml.safe_load(path.read_text(encoding="utf-8")) |
| 61 | if not isinstance(raw, dict): |
| 62 | raise ConfigError("model-labels.yaml root must be a mapping", str(path)) |
| 63 | tiers = raw.get("model_tiers") |
| 64 | if not isinstance(tiers, list) or not tiers: |
| 65 | raise ConfigError("model_tiers must be a non-empty list", str(path)) |
| 66 | ids: list[str] = [] |
| 67 | for index, entry in enumerate(tiers): |
| 68 | tier_id = validate_model_tier_entry(entry, index=index, path=str(path)) |
| 69 | ids.append(tier_id) |
| 70 | if len(ids) != len(set(ids)): |
| 71 | raise ConfigError("model_tiers[].id values must be unique", str(path)) |
| 72 | return frozenset(ids) |
| 73 | |
| 74 | |
| 75 | @lru_cache(maxsize=4) |
| 76 | def load_label_ids(kit_root: Path) -> frozenset[str]: |
| 77 | """Load ``labels[].id`` values from kit-carried policy.""" |
| 78 | path = kit_root / "policy" / "model-labels.yaml" |
| 79 | if not path.is_file(): |
| 80 | raise ConfigError("model labels registry missing", str(path)) |
| 81 | raw = yaml.safe_load(path.read_text(encoding="utf-8")) |
| 82 | if not isinstance(raw, dict): |
| 83 | raise ConfigError("model-labels.yaml root must be a mapping", str(path)) |
| 84 | labels = raw.get("labels") |
| 85 | if not isinstance(labels, list) or not labels: |
| 86 | raise ConfigError("labels must be a non-empty list", str(path)) |
| 87 | ids: list[str] = [] |
| 88 | for index, entry in enumerate(labels): |
| 89 | if not isinstance(entry, dict): |
| 90 | raise ConfigError(f"labels[{index}] must be a mapping", str(path)) |
| 91 | label_id = entry.get("id") |
| 92 | if not isinstance(label_id, str) or not label_id.strip(): |
| 93 | raise ConfigError(f"labels[{index}].id must be a non-empty string", str(path)) |
| 94 | ids.append(label_id.strip()) |
| 95 | return frozenset(ids) |
| 96 | |
| 97 | |
| 98 | @lru_cache(maxsize=4) |
| 99 | def load_model_tier_cost_bands(kit_root: Path, *, fail_closed: bool = False) -> dict[str, str | None]: |
| 100 | """Load declared ``cost_class`` bands keyed by ``model_tiers[].id``. |
| 101 | |
| 102 | When ``fail_closed`` is True, malformed ``cost_class`` values raise |
| 103 | ``RoutingPolicyError`` with exit code ``32``. |
| 104 | """ |
| 105 | path = kit_root / "policy" / "model-labels.yaml" |
| 106 | citation = str(path) |
| 107 | if not path.is_file(): |
| 108 | message = "model tier registry missing" |
| 109 | if fail_closed: |
| 110 | raise RoutingPolicyError(message, exit_code=32, citation=citation) |
| 111 | raise ConfigError(message, citation) |
| 112 | |
| 113 | raw = yaml.safe_load(path.read_text(encoding="utf-8")) |
| 114 | if not isinstance(raw, dict): |
| 115 | message = "model-labels.yaml root must be a mapping" |
| 116 | if fail_closed: |
| 117 | raise RoutingPolicyError(message, exit_code=32, citation=citation) |
| 118 | raise ConfigError(message, citation) |
| 119 | |
| 120 | tiers = raw.get("model_tiers") |
| 121 | if not isinstance(tiers, list) or not tiers: |
| 122 | message = "model_tiers must be a non-empty list" |
| 123 | if fail_closed: |
| 124 | raise RoutingPolicyError(message, exit_code=32, citation=citation) |
| 125 | raise ConfigError(message, citation) |
| 126 | |
| 127 | bands: dict[str, str | None] = {} |
| 128 | for index, entry in enumerate(tiers): |
| 129 | prefix = f"model_tiers[{index}]" |
| 130 | if not isinstance(entry, dict): |
| 131 | message = f"{prefix} must be a mapping" |
| 132 | if fail_closed: |
| 133 | raise RoutingPolicyError(message, exit_code=32, citation=citation) |
| 134 | raise ConfigError(message, citation) |
| 135 | tier_id = entry.get("id") |
| 136 | if not isinstance(tier_id, str) or not tier_id.strip(): |
| 137 | message = f"{prefix}.id must be a non-empty string" |
| 138 | if fail_closed: |
| 139 | raise RoutingPolicyError(message, exit_code=32, citation=citation) |
| 140 | raise ConfigError(message, citation) |
| 141 | cost_class = entry.get("cost_class") |
| 142 | _validate_cost_class_value( |
| 143 | cost_class, |
| 144 | prefix=prefix, |
| 145 | path=citation, |
| 146 | fail_closed=fail_closed, |
| 147 | ) |
| 148 | bands[tier_id.strip()] = cost_class if cost_class is not None else None |
| 149 | return bands |
| 150 | |
| 151 | |
| 152 | def validate_model_tier_entry(entry: object, *, index: int, path: str) -> str: |
| 153 | """Validate one ``model_tiers`` entry; return its ``id``.""" |
| 154 | prefix = f"model_tiers[{index}]" |
| 155 | if not isinstance(entry, dict): |
| 156 | raise ConfigError(f"{prefix} must be a mapping", path) |
| 157 | extra = set(entry) - MODEL_TIER_ENTRY_KEYS |
| 158 | if extra: |
| 159 | raise ConfigError(f"unknown {prefix} keys: {sorted(extra)}", path) |
| 160 | tier_id = entry.get("id") |
| 161 | if not isinstance(tier_id, str) or not tier_id.strip(): |
| 162 | raise ConfigError(f"{prefix}.id must be a non-empty string", path) |
| 163 | if not KEBAB_ID_RE.match(tier_id): |
| 164 | raise ConfigError(f"{prefix}.id must be lowercase kebab-case", path) |
| 165 | if is_vendor_slug(tier_id): |
| 166 | raise ConfigError(f"{prefix}.id must not be a vendor slug", path) |
| 167 | for field in ("display", "meaning"): |
| 168 | value = entry.get(field) |
| 169 | if not isinstance(value, str) or not value.strip(): |
| 170 | raise ConfigError(f"{prefix}.{field} must be a non-empty string", path) |
| 171 | hint = entry.get("cursor_model_hint") |
| 172 | if hint is not None: |
| 173 | if not isinstance(hint, str) or not hint.strip(): |
| 174 | raise ConfigError(f"{prefix}.cursor_model_hint must be a non-empty string", path) |
| 175 | if is_vendor_slug(hint): |
| 176 | raise ConfigError(f"{prefix}.cursor_model_hint must not contain vendor slugs", path) |
| 177 | for field in ("display", "meaning"): |
| 178 | if is_vendor_slug(entry[field]): |
| 179 | raise ConfigError(f"{prefix}.{field} must not contain vendor slugs", path) |
| 180 | _validate_cost_class_value( |
| 181 | entry.get("cost_class"), |
| 182 | prefix=prefix, |
| 183 | path=path, |
| 184 | fail_closed=False, |
| 185 | ) |
| 186 | return tier_id |
| 187 | |
| 188 | |
| 189 | def validate_model_tiers_document(raw: object, *, path: str) -> frozenset[str]: |
| 190 | """Validate a full model-labels document's ``model_tiers`` section.""" |
| 191 | if not isinstance(raw, dict): |
| 192 | raise ConfigError("model-labels.yaml root must be a mapping", path) |
| 193 | tiers = raw.get("model_tiers") |
| 194 | if not isinstance(tiers, list) or not tiers: |
| 195 | raise ConfigError("model_tiers must be a non-empty list", path) |
| 196 | ids: list[str] = [] |
| 197 | for index, entry in enumerate(tiers): |
| 198 | ids.append(validate_model_tier_entry(entry, index=index, path=path)) |
| 199 | if len(ids) != len(set(ids)): |
| 200 | raise ConfigError("model_tiers[].id values must be unique", path) |
| 201 | return frozenset(ids) |
| 202 | |
| 203 | |
| 204 | def allowed_model_tier_ids(kit_root: Path) -> frozenset[str]: |
| 205 | """Return ``model_tiers`` ids plus the reserved ``human`` terminal.""" |
| 206 | return load_model_tier_ids(kit_root) | {HUMAN_TIER} |