feat(evaluators): gateway routing based on AC policies

evaluators/contrib/budget/src/agent_control_evaluator_budget/budget/__init__.py

@@ -3,6 +3,7 @@
 from agent_control_evaluator_budget.budget.config import (
     BudgetEvaluatorConfig,
     BudgetLimitRule,
+    GatewayRoutingConfig,
     ModelPricing,
 )
 from agent_control_evaluator_budget.budget.evaluator import BudgetEvaluator
@@ -17,6 +18,7 @@
     "BudgetEvaluator",
     "BudgetEvaluatorConfig",
     "BudgetLimitRule",
+    "GatewayRoutingConfig",
     "BudgetSnapshot",
     "BudgetStore",
     "InMemoryBudgetStore",

evaluators/contrib/budget/src/agent_control_evaluator_budget/budget/config.py

@@ -5,6 +5,7 @@
 from typing import Literal
 
 from agent_control_evaluators._base import EvaluatorConfig
+from agent_control_models import GatewayModelTier
 from pydantic import Field, field_validator, model_validator
 
 # ---------------------------------------------------------------------------
@@ -68,6 +69,25 @@ def validate_window_seconds(cls, v: int | None) -> int | None:
         return v
 
 
+class GatewayRoutingConfig(EvaluatorConfig):
+    """Optional model-tier routing behavior stored on a budget control."""
+
+    tiers: dict[str, GatewayModelTier] = Field(min_length=1)
+    task_tiers: dict[str, str] = Field(min_length=1)
+    threshold: float = Field(default=0.8, gt=0.0, le=1.0)
+    threshold_tier: str | None = None
+
+    @model_validator(mode="after")
+    def validate_tier_references(self) -> GatewayRoutingConfig:
+        configured = set(self.tiers)
+        unknown = set(self.task_tiers.values()) - configured
+        if unknown:
+            raise ValueError(f"task_tiers references unknown tiers: {sorted(unknown)}")
+        if self.threshold_tier is not None and self.threshold_tier not in configured:
+            raise ValueError(f"threshold_tier references unknown tier: {self.threshold_tier}")
+        return self
+
+
 class BudgetEvaluatorConfig(EvaluatorConfig):
     """Configuration for the budget evaluator.
 
@@ -109,6 +129,7 @@ class BudgetEvaluatorConfig(EvaluatorConfig):
     token_path: str | None = None
     model_path: str | None = None
     metadata_paths: dict[str, str] = Field(default_factory=dict)
+    gateway_routing: GatewayRoutingConfig | None = None
 
     @model_validator(mode="after")
     def require_pricing_for_cost_rules(self) -> BudgetEvaluatorConfig:
@@ -117,4 +138,8 @@ def require_pricing_for_cost_rules(self) -> BudgetEvaluatorConfig:
             raise ValueError('pricing is required when any rule uses limit_unit="usd_cents"')
         if has_cost_rule and not (self.model_path or "").strip():
             raise ValueError('model_path is required when any rule uses limit_unit="usd_cents"')
+        if self.gateway_routing is not None and not any(
+            rule.limit_unit == "tokens" for rule in self.limits
+        ):
+            raise ValueError("gateway_routing requires at least one token budget limit")
         return self

evaluators/contrib/budget/src/agent_control_evaluator_budget/budget/evaluator.py

@@ -20,9 +20,9 @@
 
 from agent_control_evaluators._base import Evaluator, EvaluatorMetadata
 from agent_control_evaluators._registry import register_evaluator
-from agent_control_models import EvaluatorResult
+from agent_control_models import EvaluatorResult, GatewayModelCandidate
 
-from .config import BudgetEvaluatorConfig, ModelPricing
+from .config import BudgetEvaluatorConfig, GatewayRoutingConfig, ModelPricing
 from .memory_store import InMemoryBudgetStore, _scope_matches
 from .store import BudgetStore
 
@@ -174,6 +174,92 @@ def _extract_metadata(data: Any, metadata_paths: dict[str, str]) -> dict[str, st
     return result
 
 
+def _gateway_route_metadata(
+    data: Any,
+    routing: GatewayRoutingConfig | None,
+    snapshots: list[dict[str, Any]],
+    *,
+    over_limit: bool,
+) -> dict[str, Any] | None:
+    """Select a configured candidate using task tier and accumulated token use."""
+    if routing is None:
+        return None
+    request_data = data if isinstance(data, dict) else {}
+    task_type = request_data.get("task_type", "default")
+    if not isinstance(task_type, str):
+        task_type = "default"
+    budget_snapshot = next(
+        (snapshot for snapshot in snapshots if snapshot.get("limit_unit") == "tokens"), {}
+    )
+    used_tokens = int(budget_snapshot.get("spent_tokens", 0))
+    limit_tokens = int(budget_snapshot.get("limit", 0))
+    utilization = float(budget_snapshot.get("utilization", 0.0))
+    threshold_reached = utilization >= routing.threshold
+    task_tier = routing.task_tiers.get(task_type, routing.task_tiers.get("default"))
+    selected_tier = (
+        routing.threshold_tier
+        if threshold_reached and routing.threshold_tier is not None
+        else task_tier
+    )
+    reason = ""
+    candidate: GatewayModelCandidate | None = None
+    if over_limit:
+        reason = "The configured token budget is exhausted."
+    elif selected_tier is None:
+        reason = f"No model tier is configured for task type '{task_type}'."
+    else:
+        raw_candidates = request_data.get("candidate_models", [])
+        candidates: set[tuple[str, str, str]] = set()
+        legacy_model = request_data.get("model")
+        if isinstance(raw_candidates, list):
+            for raw_candidate in raw_candidates:
+                if isinstance(raw_candidate, dict):
+                    try:
+                        parsed = GatewayModelCandidate.model_validate(raw_candidate)
+                    except ValueError:
+                        continue
+                    candidates.add((parsed.provider, parsed.model, parsed.protocol))
+        if not candidates and isinstance(legacy_model, str):
+            candidates = {
+                (configured.provider, configured.model, configured.protocol)
+                for configured in routing.tiers[selected_tier].models
+                if configured.model == legacy_model
+            }
+        candidate = next(
+            (
+                configured
+                for configured in routing.tiers[selected_tier].models
+                if (configured.provider, configured.model, configured.protocol) in candidates
+                and (not isinstance(legacy_model, str) or configured.model == legacy_model)
+            ),
+            None,
+        )
+        reason = (
+            f"Budget threshold reached; routed to the '{selected_tier}' tier."
+            if threshold_reached
+            else f"Task type '{task_type}' routed to the '{selected_tier}' tier."
+        ) if candidate is not None else (
+            f"No compatible model candidate is available in tier '{selected_tier}'."
+        )
+    return {
+        "allowed": candidate is not None and not over_limit,
+        "provider": candidate.provider if candidate is not None and not over_limit else None,
+        "model": candidate.model if candidate is not None and not over_limit else None,
+        "protocol": candidate.protocol if candidate is not None and not over_limit else None,
+        "task_type": task_type,
+        "tier": selected_tier,
+        "budget": {
+            "used_tokens": used_tokens,
+            "limit_tokens": limit_tokens,
+            "utilization": min(utilization, 1.0),
+            "threshold": routing.threshold,
+            "threshold_reached": threshold_reached,
+            "over_limit": over_limit,
+        },
+        "reason": reason,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Evaluator
 # ---------------------------------------------------------------------------
@@ -281,6 +367,9 @@ async def evaluate(self, data: Any) -> EvaluatorResult:
 
         if breached:
             first = breached[0]
+            route_metadata = _gateway_route_metadata(
+                data, self.config.gateway_routing, all_snaps, over_limit=True
+            )
             return EvaluatorResult(
                 matched=True,
                 confidence=1.0,
@@ -291,19 +380,31 @@ async def evaluate(self, data: Any) -> EvaluatorResult:
                     "input_tokens": input_tokens,
                     "output_tokens": output_tokens,
                     "cost": round(cost, 6),
+                    **({"gateway_route": route_metadata} if route_metadata is not None else {}),
                 },
             )
 
         max_util = max((s["utilization"] for s in all_snaps), default=0.0)
+        route_metadata = _gateway_route_metadata(
+            data, self.config.gateway_routing, all_snaps, over_limit=False
+        )
         return EvaluatorResult(
-            matched=False,
+            matched=(
+                self.config.gateway_routing is not None
+                or bool(route_metadata and route_metadata.get("allowed"))
+            ),
             confidence=1.0,
-            message=f"Within budget (utilization={max_util:.0%})",
+            message=(
+                str(route_metadata.get("reason"))
+                if route_metadata
+                else f"Within budget (utilization={max_util:.0%})"
+            ),
             metadata={
                 "all_snapshots": all_snaps,
                 "input_tokens": input_tokens,
                 "output_tokens": output_tokens,
                 "cost": round(cost, 6),
                 "max_utilization": round(max_util, 4),
+                **({"gateway_route": route_metadata} if route_metadata is not None else {}),
             },
         )