| 262 | |
| 263 | |
| 264 | def apply_quality_guards( |
| 265 | candidates: list[str], |
| 266 | *, |
| 267 | mode: RoutingMode, |
| 268 | tier: Tier, |
| 269 | lane: CapabilityLane, |
| 270 | capabilities: dict[str, ModelCapabilities], |
| 271 | continuity_floor: ServedQuality | None = None, |
| 272 | step_risk: str = "normal", |
| 273 | agent_pressure: float = 0.0, |
| 274 | is_agentic: bool = False, |
| 275 | has_tool_results: bool = False, |
| 276 | ) -> QualityGuardResult: |
| 277 | quality_by_model = { |
| 278 | model: model_served_quality(model, lane, capabilities.get(model)) |
| 279 | for model in candidates |
| 280 | } |
| 281 | target = target_served_quality(mode, tier) |
| 282 | normalized_step_risk = str(step_risk or "normal").strip().lower() |
| 283 | floor = minimum_served_quality(mode, tier) |
| 284 | if lane is CapabilityLane.REASONING and tier is Tier.COMPLEX: |
| 285 | lane_floor = ServedQuality.PREMIUM |
| 286 | else: |
| 287 | lane_floor = None |
| 288 | if normalized_step_risk == "high": |
| 289 | risk_floor = ServedQuality.BALANCED |
| 290 | else: |
| 291 | risk_floor = None |
| 292 | hard_continuity_floor = continuity_floor if mode is RoutingMode.BEST else None |
| 293 | effective_floor = stronger_quality( |
| 294 | stronger_quality(stronger_quality(floor, lane_floor), risk_floor), |
| 295 | hard_continuity_floor, |
| 296 | ) or floor |
| 297 | preferred_threshold = ( |
| 298 | stronger_quality(stronger_quality(target, effective_floor), hard_continuity_floor) |
| 299 | or target |
| 300 | ) |
| 301 | notes: list[str] = [ |
| 302 | f"lane={lane.value}", |
| 303 | f"served-quality-target={target.value}", |
| 304 | f"served-quality-floor={effective_floor.value}", |
| 305 | ] |
| 306 | if normalized_step_risk != "normal": |
| 307 | notes.append(f"step-risk={normalized_step_risk}") |
| 308 | if agent_pressure >= 0.35: |
| 309 | notes.append(f"agent-pressure={agent_pressure:.2f}") |
| 310 | if risk_floor is not None and normalized_step_risk == "high": |
| 311 | notes.append(f"step-risk-floor={risk_floor.value}") |
| 312 | elif risk_floor is not None: |
| 313 | notes.append(f"agent-pressure-floor={risk_floor.value}") |
| 314 | if lane_floor is not None: |
| 315 | notes.append(f"lane-floor={lane_floor.value}") |
| 316 | if continuity_floor is not None and hard_continuity_floor is None: |
| 317 | notes.append(f"continuity-soft={continuity_floor.value}") |
| 318 | exact_quality = stronger_quality(preferred_threshold, effective_floor) or target |
| 319 | exact = [ |
| 320 | model for model in candidates |
| 321 | if quality_by_model[model] is exact_quality |