aaronrene / overseer-kit public
policy.py python
266 lines 10.0 KB
Raw
sha256:c07f2f34a0db9f43fe866f157d1935322008545ff8940c06c7c921eb219c55ab NXP-b DONE: independent BV-r2 pass + ISR (SD-17) Human minor ⚠ breaking 18 hours ago
1 """``policy/model-routing.yaml`` load and validation (§PR.4)."""
2
3 from __future__ import annotations
4
5 from pathlib import Path
6 from typing import Any
7
8 import yaml
9
10 from tools.freeze_reviewer.labels import is_vendor_slug
11 from tools.model_routing.labels import HUMAN_TIER, RoutingPolicyError, allowed_model_tier_ids
12 from tools.model_routing.types import PolicyValidationResult, RouteEntry, RouteSelector, RoutingPolicy
13
14 ROUTING_POLICY_VERSION = 1
15 TOP_LEVEL_KEYS = frozenset({"version", "defaults", "routes"})
16 DEFAULTS_KEYS = frozenset({"model_tier", "fallback"})
17 ROUTE_KEYS = frozenset({"id", "when", "model_tier", "fallback"})
18 WHEN_KEYS = frozenset({"position", "phase_tier", "gate"})
19 KIT_GATE_VALUES = frozenset({"freeze_review", "build_verification", "default"})
20
21
22 def load_routing_policy_text(text: str, *, kit_root: Path, citation: str) -> RoutingPolicy:
23 """Parse routing policy YAML from text."""
24 try:
25 raw = yaml.safe_load(text)
26 except yaml.YAMLError as exc:
27 raise RoutingPolicyError(
28 f"unparseable routing policy YAML: {exc}",
29 citation=citation,
30 ) from exc
31 return parse_routing_policy(raw, kit_root=kit_root, citation=citation)
32
33
34 def load_routing_policy(path: Path, *, kit_root: Path) -> RoutingPolicy:
35 """Read and parse a routing policy file; exit ``31`` when missing/unreadable."""
36 citation = str(path)
37 if not path.is_file():
38 raise RoutingPolicyError(
39 "routing policy file missing or unreadable",
40 exit_code=31,
41 citation=citation,
42 )
43 try:
44 text = path.read_text(encoding="utf-8")
45 except OSError as exc:
46 raise RoutingPolicyError(
47 f"routing policy file missing or unreadable: {exc}",
48 exit_code=31,
49 citation=citation,
50 ) from exc
51 return load_routing_policy_text(text, kit_root=kit_root, citation=citation)
52
53
54 def parse_routing_policy(raw: Any, *, kit_root: Path, citation: str) -> RoutingPolicy:
55 """Parse a routing policy mapping; raise ``RoutingPolicyError`` on violation."""
56 if not isinstance(raw, dict):
57 raise RoutingPolicyError("routing policy root must be a mapping", citation=citation)
58
59 extra = set(raw) - TOP_LEVEL_KEYS
60 if extra:
61 raise RoutingPolicyError(
62 f"unknown top-level routing policy keys: {sorted(extra)}",
63 citation=citation,
64 )
65
66 version = raw.get("version")
67 if version != ROUTING_POLICY_VERSION:
68 raise RoutingPolicyError(
69 f"unsupported routing policy version {version!r} (supported: {ROUTING_POLICY_VERSION})",
70 citation=citation,
71 )
72
73 if "defaults" not in raw:
74 raise RoutingPolicyError("routing policy missing mandatory defaults", citation=citation)
75
76 allowed_tiers = allowed_model_tier_ids(kit_root)
77 defaults_model_tier, defaults_fallback = _parse_route_target(
78 raw["defaults"],
79 field_prefix="defaults",
80 citation=citation,
81 allowed_tiers=allowed_tiers,
82 )
83
84 routes_raw = raw.get("routes", [])
85 if routes_raw is None:
86 routes_raw = []
87 if not isinstance(routes_raw, list):
88 raise RoutingPolicyError("routes must be a list", citation=citation)
89
90 routes: list[RouteEntry] = []
91 seen_ids: set[str] = set()
92 for index, item in enumerate(routes_raw):
93 prefix = f"routes[{index}]"
94 if not isinstance(item, dict):
95 raise RoutingPolicyError(f"{prefix} must be a mapping", citation=citation)
96 route_extra = set(item) - ROUTE_KEYS
97 if route_extra:
98 raise RoutingPolicyError(
99 f"unknown {prefix} keys: {sorted(route_extra)}",
100 citation=citation,
101 )
102 route_id = item.get("id")
103 if not isinstance(route_id, str) or not route_id.strip():
104 raise RoutingPolicyError(f"{prefix}.id must be a non-empty string", citation=citation)
105 if route_id in seen_ids:
106 raise RoutingPolicyError(f"duplicate route.id {route_id!r}", citation=citation)
107 _reject_vendor_slug(route_id, f"{prefix}.id", citation)
108 seen_ids.add(route_id)
109
110 when = _parse_when(item.get("when"), prefix=prefix, citation=citation)
111 model_tier, fallback = _parse_route_target(
112 item,
113 field_prefix=prefix,
114 citation=citation,
115 allowed_tiers=allowed_tiers,
116 )
117 routes.append(
118 RouteEntry(
119 id=route_id,
120 when=when,
121 model_tier=model_tier,
122 fallback=fallback,
123 )
124 )
125
126 _scan_strings_for_vendor_slugs(raw, citation=citation)
127 return RoutingPolicy(
128 version=ROUTING_POLICY_VERSION,
129 defaults_model_tier=defaults_model_tier,
130 defaults_fallback=defaults_fallback,
131 routes=tuple(routes),
132 )
133
134
135 def validate_routing_policy(path: Path, *, kit_root: Path) -> PolicyValidationResult:
136 """Validate routing policy; return first violation with citation."""
137 try:
138 load_routing_policy(path, kit_root=kit_root)
139 except RoutingPolicyError as exc:
140 return PolicyValidationResult(valid=False, violation=_format_violation(exc))
141 return PolicyValidationResult(valid=True)
142
143
144 def _format_violation(exc: RoutingPolicyError) -> str:
145 if exc.citation:
146 return f"{exc.citation}: {exc.message}"
147 return exc.message
148
149
150 def _parse_when(raw_when: Any, *, prefix: str, citation: str) -> RouteSelector:
151 if raw_when is None:
152 return RouteSelector()
153 if not isinstance(raw_when, dict):
154 raise RoutingPolicyError(f"{prefix}.when must be a mapping", citation=citation)
155 when_extra = set(raw_when) - WHEN_KEYS
156 if when_extra:
157 raise RoutingPolicyError(
158 f"unknown {prefix}.when keys: {sorted(when_extra)}",
159 citation=citation,
160 )
161 position = _optional_selector_value(raw_when, "position", prefix=prefix, citation=citation)
162 phase_tier = _optional_selector_value(raw_when, "phase_tier", prefix=prefix, citation=citation)
163 gate = _optional_selector_value(raw_when, "gate", prefix=prefix, citation=citation)
164 if gate is not None and gate not in KIT_GATE_VALUES:
165 raise RoutingPolicyError(
166 f"{prefix}.when.gate must be freeze_review|build_verification|default",
167 citation=citation,
168 )
169 return RouteSelector(position=position, phase_tier=phase_tier, gate=gate)
170
171
172 def _optional_selector_value(
173 mapping: dict[str, Any],
174 field: str,
175 *,
176 prefix: str,
177 citation: str,
178 ) -> str | None:
179 if field not in mapping:
180 return None
181 value = mapping[field]
182 if not isinstance(value, str) or not value.strip():
183 raise RoutingPolicyError(f"{prefix}.when.{field} must be a non-empty string", citation=citation)
184 return value
185
186
187 def _parse_route_target(
188 mapping: dict[str, Any],
189 *,
190 field_prefix: str,
191 citation: str,
192 allowed_tiers: frozenset[str],
193 ) -> tuple[str, tuple[str, ...]]:
194 if field_prefix == "defaults":
195 keys = DEFAULTS_KEYS
196 extra = set(mapping) - DEFAULTS_KEYS
197 if extra:
198 raise RoutingPolicyError(
199 f"unknown defaults keys: {sorted(extra)}",
200 citation=citation,
201 )
202 else:
203 keys = frozenset({"model_tier", "fallback"})
204
205 if "model_tier" not in mapping:
206 raise RoutingPolicyError(f"{field_prefix}.model_tier is required", citation=citation)
207 model_tier = mapping["model_tier"]
208 if not isinstance(model_tier, str) or not model_tier.strip():
209 raise RoutingPolicyError(f"{field_prefix}.model_tier must be a non-empty string", citation=citation)
210 if model_tier not in allowed_tiers:
211 raise RoutingPolicyError(
212 f"{field_prefix}.model_tier {model_tier!r} is not in model_tiers",
213 citation=citation,
214 )
215 _reject_vendor_slug(model_tier, f"{field_prefix}.model_tier", citation)
216
217 fallback_raw = mapping.get("fallback")
218 if not isinstance(fallback_raw, list) or not fallback_raw:
219 raise RoutingPolicyError(f"{field_prefix}.fallback must be a non-empty list", citation=citation)
220 fallback: list[str] = []
221 for index, item in enumerate(fallback_raw):
222 if not isinstance(item, str) or not item.strip():
223 raise RoutingPolicyError(
224 f"{field_prefix}.fallback[{index}] must be a non-empty string",
225 citation=citation,
226 )
227 if item not in allowed_tiers:
228 raise RoutingPolicyError(
229 f"{field_prefix}.fallback[{index}] {item!r} is not in model_tiers",
230 citation=citation,
231 )
232 _reject_vendor_slug(item, f"{field_prefix}.fallback[{index}]", citation)
233 fallback.append(item)
234
235 if fallback[0] != model_tier:
236 raise RoutingPolicyError(
237 f"{field_prefix}.fallback[0] must equal {field_prefix}.model_tier",
238 citation=citation,
239 )
240 if fallback[-1] != HUMAN_TIER:
241 raise RoutingPolicyError(
242 f"{field_prefix}.fallback must terminate in {HUMAN_TIER!r}",
243 citation=citation,
244 )
245 return model_tier, tuple(fallback)
246
247
248 def _reject_vendor_slug(value: str, field: str, citation: str) -> None:
249 if is_vendor_slug(value):
250 raise RoutingPolicyError(f"{field} must not contain vendor slugs", citation=citation)
251
252
253 def _scan_strings_for_vendor_slugs(node: Any, *, citation: str, path: str = "") -> None:
254 """Reject vendor slugs embedded in any string value (including comments rendered as values)."""
255 if isinstance(node, dict):
256 for key, value in node.items():
257 child = f"{path}.{key}" if path else str(key)
258 if isinstance(key, str):
259 _reject_vendor_slug(key, child, citation)
260 _scan_strings_for_vendor_slugs(value, citation=citation, path=child)
261 elif isinstance(node, list):
262 for index, value in enumerate(node):
263 _scan_strings_for_vendor_slugs(value, citation=citation, path=f"{path}[{index}]")
264 elif isinstance(node, str):
265 if is_vendor_slug(node):
266 raise RoutingPolicyError(f"{path or 'policy'} must not contain vendor slugs", citation=citation)
File History 1 commit
sha256:c07f2f34a0db9f43fe866f157d1935322008545ff8940c06c7c921eb219c55ab NXP-b DONE: independent BV-r2 pass + ISR (SD-17) Human minor ⚠ 18 hours ago