6e6d02fac6
Implements the bugs_to_fix/capability-gate-followups.md plan (#1, #2, #3, #4, #5, #6). - #1: local vision fallback honors require_json_mode - an image + json_object request now falls through to the 422 naming the missing capability instead of returning prose that breaks the json contract - #2: the pinned-model capability check now honors cfg.routing.require_vision / require_json_mode, so a pin and routed traffic face the same gate - #3: drop 'image' modality from deepseek-v4-flash (supports_vision = 0) - #4: README hard-filter count Four -> Six - #5: extract routing.capability_gate_reason() shared by rejection_reason and the pinned check, so the flag rule cannot drift between the two - #6: single iter_image_url_values() generator replaces four image-part traversals; _run_local_vision computes count/bytes once Adds 9 tests (capability_gate_reason unit coverage, json-mode-gated local fallback, vision gate off for pins). Full suite: 416 passing.
501 lines
20 KiB
Python
501 lines
20 KiB
Python
"""Tests for routing.py — hard filters and weighted ranking.
|
|
|
|
The filters are the part that disqualifies outright, so each one gets a case
|
|
proving it rejects and a case proving it does not over-reject.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from routing import (
|
|
capability_gate_reason,
|
|
is_eligible,
|
|
rank_candidates,
|
|
rejection_reason,
|
|
select_candidates,
|
|
)
|
|
|
|
|
|
|
|
def _row(**overrides) -> dict:
|
|
"""A routable model row; override one field per test."""
|
|
row = {
|
|
"model_id": "m",
|
|
"provider": "neuralwatt",
|
|
"tier": 2,
|
|
"cost": 1.0,
|
|
"energy": 1.0e-5,
|
|
"effective_context_window": 100_000,
|
|
"availability": "active",
|
|
"deprecated": 0,
|
|
"access_level": "public",
|
|
"latency_class": "standard",
|
|
"reasoning_mode": "default",
|
|
"context_variant": "full",
|
|
"supports_vision": 1,
|
|
"supports_json_mode": 1,
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
def _eligible(row, **overrides) -> bool:
|
|
kwargs = {
|
|
"required_context_tokens": 10_000,
|
|
"required_tier": 2,
|
|
"latency_tolerance": "interactive",
|
|
"allowed_access_levels": ["public"],
|
|
"exclude_stale": True,
|
|
"exclude_deprecated": True,
|
|
}
|
|
kwargs.update(overrides)
|
|
return is_eligible(row, **kwargs)
|
|
|
|
|
|
# --- context window -------------------------------------------------------
|
|
|
|
def test_context_window_too_small_is_rejected():
|
|
assert _eligible(_row(effective_context_window=5_000)) is False
|
|
|
|
|
|
def test_context_window_exactly_equal_is_accepted():
|
|
# The filter is >=, so a model that exactly fits is still a candidate
|
|
assert _eligible(_row(effective_context_window=10_000)) is True
|
|
|
|
|
|
def test_null_context_window_is_rejected():
|
|
# Given: a row whose window never got derived. An unknown window cannot be
|
|
# shown to fit, and truncating mid-task is worse than routing elsewhere.
|
|
assert _eligible(_row(effective_context_window=None)) is False
|
|
|
|
|
|
# --- tier floor -----------------------------------------------------------
|
|
|
|
def test_tier_below_required_is_rejected():
|
|
assert _eligible(_row(tier=1), required_tier=2) is False
|
|
|
|
|
|
def test_tier_above_required_is_accepted():
|
|
# tier is a floor, not an equality match
|
|
assert _eligible(_row(tier=3), required_tier=2) is True
|
|
|
|
|
|
def test_null_tier_is_rejected():
|
|
assert _eligible(_row(tier=None)) is False
|
|
|
|
|
|
# --- freshness ------------------------------------------------------------
|
|
|
|
def test_stale_row_is_rejected_when_configured():
|
|
assert _eligible(_row(availability="stale")) is False
|
|
|
|
|
|
def test_stale_row_is_kept_when_not_excluded():
|
|
assert _eligible(_row(availability="stale"), exclude_stale=False) is True
|
|
|
|
|
|
def test_deprecated_row_is_rejected():
|
|
assert _eligible(_row(availability="deprecated")) is False
|
|
assert _eligible(_row(deprecated=1)) is False
|
|
|
|
|
|
# --- access gating --------------------------------------------------------
|
|
|
|
def test_grant_gated_rows_are_rejected_by_default():
|
|
# Given: the glm-5.2-short* rows, which are private preview. Routing to
|
|
# one earns a 403 at dispatch, so they are excluded before scoring.
|
|
assert _eligible(_row(access_level="preview")) is False
|
|
assert _eligible(_row(access_level="canary")) is False
|
|
|
|
|
|
def test_gated_rows_are_admitted_when_the_account_holds_the_grant():
|
|
assert (
|
|
_eligible(_row(access_level="preview"), allowed_access_levels=["public", "preview"])
|
|
is True
|
|
)
|
|
|
|
|
|
# --- latency tolerance ----------------------------------------------------
|
|
|
|
def test_flex_row_is_rejected_for_interactive_work():
|
|
# Given: a flex row, held server-side during peak until capacity frees up
|
|
assert _eligible(_row(latency_class="flex"), latency_tolerance="interactive") is False
|
|
|
|
|
|
def test_flex_row_is_admitted_for_batch_work():
|
|
assert _eligible(_row(latency_class="flex"), latency_tolerance="batch") is True
|
|
|
|
|
|
def test_standard_row_is_admitted_for_batch_work():
|
|
# Batch tolerates flex; it does not require it
|
|
assert _eligible(_row(latency_class="standard"), latency_tolerance="batch") is True
|
|
|
|
|
|
# --- select_candidates ----------------------------------------------------
|
|
|
|
def test_select_candidates_filters_and_preserves_order():
|
|
rows = [
|
|
_row(model_id="keep-1"),
|
|
_row(model_id="drop-flex", latency_class="flex"),
|
|
_row(model_id="keep-2", tier=3),
|
|
_row(model_id="drop-gated", access_level="preview"),
|
|
]
|
|
selected = select_candidates(
|
|
rows,
|
|
required_context_tokens=10_000,
|
|
required_tier=2,
|
|
latency_tolerance="interactive",
|
|
allowed_access_levels=["public"],
|
|
exclude_stale=True,
|
|
exclude_deprecated=True,
|
|
)
|
|
assert [r["model_id"] for r in selected] == ["keep-1", "keep-2"]
|
|
|
|
|
|
# --- measured cost --------------------------------------------------------
|
|
|
|
# --- ranking: quality first, cheapest among equals -------------------------
|
|
|
|
def test_a_real_quality_gap_decides_outright():
|
|
# tool_use_agentic spans 0.67 across the catalog; a gap that size must
|
|
# beat any cost saving
|
|
rows = [_row(model_id="cheap-bad", cost=1e-6, proficiency=0.33),
|
|
_row(model_id="dear-good", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows)[0]["model_id"] == "dear-good"
|
|
|
|
|
|
def test_within_tolerance_the_cheaper_model_wins():
|
|
# 0.02 apart on 2-3 samples is sampling noise, not a quality difference.
|
|
# Paying 100x for it would be buying noise.
|
|
rows = [_row(model_id="cheap", cost=1e-5, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows, quality_tolerance=0.1)[0]["model_id"] == "cheap"
|
|
|
|
|
|
def test_narrowing_the_tolerance_makes_small_gaps_count():
|
|
# As samples accumulate and confidence rises, the band should shrink
|
|
rows = [_row(model_id="cheap", cost=1e-5, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows, quality_tolerance=0.001)[0]["model_id"] == "dear"
|
|
|
|
|
|
def test_eco_no_longer_influences_the_decision():
|
|
# Carbon is still logged; it is simply not what this router optimizes
|
|
rows = [_row(model_id="clean", cost=1e-3, proficiency=0.5, eco=1e-6),
|
|
_row(model_id="dirty", cost=1e-5, proficiency=0.5, eco=1e9)]
|
|
assert rank_candidates(rows)[0]["model_id"] == "dirty"
|
|
|
|
|
|
def test_missing_proficiency_is_neutral_not_penalized():
|
|
ranked = rank_candidates([_row()])
|
|
assert ranked[0]["proficiency_score"] == 0.5
|
|
|
|
|
|
def test_unknown_cost_does_not_disqualify(db_free=None):
|
|
# A model never swept must stay pickable, or it can never acquire a
|
|
# measurement — the same trap the neutral-0.5 default avoids
|
|
ranked = rank_candidates([_row(model_id="unswept", cost=None)])
|
|
assert ranked[0]["model_id"] == "unswept"
|
|
|
|
|
|
def test_ties_break_deterministically():
|
|
rows = [_row(model_id="glm-b"), _row(model_id="glm-a"), _row(model_id="glm-c")]
|
|
first = [r["model_id"] for r in rank_candidates(rows)]
|
|
second = [r["model_id"] for r in rank_candidates(list(reversed(rows)))]
|
|
assert first == ["glm-a", "glm-b", "glm-c"] == second
|
|
|
|
|
|
def test_empty_candidate_set_ranks_to_empty():
|
|
assert rank_candidates([]) == []
|
|
|
|
|
|
# --- the budget ceiling: the quota mandate as a guarantee ------------------
|
|
|
|
def test_ceiling_excludes_models_over_budget():
|
|
# Denominated in kWh: the plan is a fixed quota, and a quota is a wall you
|
|
# hit mid-task rather than a bill that accrues
|
|
rows = [_row(model_id="affordable", energy=1e-6, proficiency=0.5),
|
|
_row(model_id="expensive", energy=1e-3, proficiency=1.0)]
|
|
ranked = rank_candidates(rows, max_energy_per_request=1e-4)
|
|
assert [r["model_id"] for r in ranked] == ["affordable"]
|
|
|
|
|
|
def test_ceiling_binds_even_against_the_best_model():
|
|
# This is the point of a constraint rather than a weight: no amount of
|
|
# quality buys past the ceiling
|
|
rows = [_row(model_id="expensive", energy=1e-3, proficiency=1.0)]
|
|
assert rank_candidates(rows, max_energy_per_request=1e-6) == []
|
|
|
|
|
|
def test_no_ceiling_admits_everything():
|
|
rows = [_row(model_id="expensive", energy=1e9, proficiency=1.0)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=None)) == 1
|
|
|
|
|
|
def test_unmeasured_cost_is_admitted_under_a_ceiling():
|
|
# Excluding the unmeasured would mean a new model could never be picked
|
|
# and so could never acquire a measurement
|
|
rows = [_row(model_id="unswept", energy=None, proficiency=1.0)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=1e-9)) == 1
|
|
|
|
|
|
def test_ceiling_at_exactly_the_cost_admits():
|
|
rows = [_row(model_id="borderline", energy=1e-6, proficiency=0.5)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=1e-6)) == 1
|
|
|
|
|
|
# --- tool competence is read from the request, not guessed at -------------
|
|
|
|
def test_tools_in_the_request_exclude_a_model_that_overreaches_for_them():
|
|
# deepseek-v4-flash measures 0.33 here. The recorded failure is a
|
|
# NON-agentic prompt ("it is 1:20pm, my meeting is at 3pm, how many
|
|
# minutes?") where it called two tools instead of subtracting -- so the
|
|
# hazard is tools being available, not the task being agentic.
|
|
weak = _row(model_id="overreacher", proficiency=1.0, tool_proficiency=0.33)
|
|
strong = _row(model_id="reliable", proficiency=1.0, tool_proficiency=1.0)
|
|
assert _eligible(weak, min_tool_proficiency=0.5) is False
|
|
assert _eligible(strong, min_tool_proficiency=0.5) is True
|
|
|
|
|
|
def test_without_tools_the_same_model_is_fine():
|
|
# The filter applies only when the request carries tool definitions;
|
|
# otherwise a cheap tool-clumsy model is a perfectly good choice.
|
|
weak = _row(model_id="overreacher", tool_proficiency=0.33)
|
|
assert _eligible(weak, min_tool_proficiency=None) is True
|
|
|
|
|
|
def test_an_unmeasured_model_is_unproven_not_disqualified():
|
|
# Same principle as the tier-1 context gate: absent evidence must not
|
|
# decide anything. A model nobody has evaluated for tool use yet is
|
|
# unproven, not proven bad.
|
|
unknown = _row(model_id="unevaluated", tool_proficiency=None)
|
|
assert _eligible(unknown, min_tool_proficiency=0.5) is True
|
|
|
|
|
|
def test_the_threshold_is_exclusive_at_the_boundary():
|
|
assert _eligible(_row(tool_proficiency=0.5), min_tool_proficiency=0.5) is True
|
|
assert _eligible(_row(tool_proficiency=0.49), min_tool_proficiency=0.5) is False
|
|
|
|
|
|
def test_the_tool_filter_can_change_the_winner():
|
|
# The whole point: it is a hard filter, so it removes a candidate that
|
|
# would otherwise win on cost rather than merely penalizing it.
|
|
rows = [
|
|
_row(model_id="cheap-clumsy", proficiency=1.0, tool_proficiency=0.33,
|
|
cost_per_1m_prompt=0.1, cost_per_1m_completion=0.2),
|
|
_row(model_id="dearer-reliable", proficiency=1.0, tool_proficiency=1.0,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0),
|
|
]
|
|
no_tools = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, min_tool_proficiency=None)
|
|
with_tools = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, min_tool_proficiency=0.5)
|
|
|
|
assert rank_candidates(no_tools, prompt_tokens=1000)[0]["model_id"] == "cheap-clumsy"
|
|
assert rank_candidates(with_tools, prompt_tokens=1000)[0]["model_id"] == "dearer-reliable"
|
|
|
|
|
|
# --- capability filters ----------------------------------------------------
|
|
|
|
def test_vision_required_excludes_a_model_without_vision():
|
|
blind = _row(model_id="blind", supports_vision=0)
|
|
sighted = _row(model_id="sighted", supports_vision=1)
|
|
assert _eligible(blind, require_vision=True) is False
|
|
assert _eligible(sighted, require_vision=True) is True
|
|
|
|
|
|
def test_vision_not_required_admits_everyone():
|
|
# The gate is off by default; not asking for vision must not exclude
|
|
blind = _row(model_id="blind", supports_vision=0)
|
|
assert _eligible(blind, require_vision=False) is True
|
|
|
|
|
|
def test_unknown_vision_capability_fails_closed():
|
|
# A capability FLAG that is absent means "cannot confirm". Unlike a
|
|
# proficiency MEASUREMENT, admit-on-None would route a vision request to a
|
|
# model that might lack it -- a guaranteed provider 400.
|
|
unknown = _row(model_id="unknown", supports_vision=None)
|
|
assert _eligible(unknown, require_vision=True) is False
|
|
|
|
|
|
def test_json_mode_required_excludes_without_support():
|
|
plain = _row(model_id="plain", supports_json_mode=0)
|
|
json = _row(model_id="json", supports_json_mode=1)
|
|
assert _eligible(plain, require_json_mode=True) is False
|
|
assert _eligible(json, require_json_mode=True) is True
|
|
|
|
|
|
def test_unknown_json_mode_fails_closed():
|
|
unknown = _row(model_id="unknown", supports_json_mode=None)
|
|
assert _eligible(unknown, require_json_mode=True) is False
|
|
|
|
|
|
def test_json_mode_gate_admits_supporting_model():
|
|
assert _eligible(_row(supports_json_mode=1), require_json_mode=True) is True
|
|
|
|
|
|
def test_the_vision_filter_can_change_the_winner():
|
|
# Same shape as the tool-filter winner test: the gate is a hard filter, so
|
|
# it removes a model that would otherwise win on cost rather than merely
|
|
# penalizing it.
|
|
rows = [
|
|
_row(model_id="cheap-blind", proficiency=1.0, supports_vision=0,
|
|
cost_per_1m_prompt=0.1, cost_per_1m_completion=0.2),
|
|
_row(model_id="dearer-sighted", proficiency=1.0, supports_vision=1,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0),
|
|
]
|
|
no_images = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, require_vision=False)
|
|
with_images = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, require_vision=True)
|
|
|
|
assert rank_candidates(no_images, prompt_tokens=1000)[0]["model_id"] == "cheap-blind"
|
|
assert rank_candidates(with_images, prompt_tokens=1000)[0]["model_id"] == "dearer-sighted"
|
|
|
|
|
|
def test_rejection_reason_names_the_missing_capability():
|
|
assert _reason(_row(supports_vision=0), require_vision=True) == "vision(unsupported)"
|
|
assert _reason(_row(supports_vision=None), require_vision=True) == "vision(unknown)"
|
|
assert _reason(_row(supports_json_mode=0), require_json_mode=True) == "json_mode(unsupported)"
|
|
assert _reason(_row(supports_json_mode=None), require_json_mode=True) == "json_mode(unknown)"
|
|
|
|
|
|
def test_zero_tolerance_ranks_strictly_on_quality():
|
|
""""Never trade quality for cost" is a legitimate setting.
|
|
|
|
The validator has always accepted 0; `band()` divided by it, so every
|
|
request raised ZeroDivisionError on a config that had loaded cleanly.
|
|
"""
|
|
rows = [_row(model_id="cheap", cost=1e-9, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
|
|
assert rank_candidates(rows, quality_tolerance=0)[0]["model_id"] == "dear"
|
|
|
|
|
|
def test_zero_tolerance_still_breaks_exact_ties_on_cost():
|
|
rows = [_row(model_id="dear", cost=1e-3, proficiency=1.00),
|
|
_row(model_id="cheap", cost=1e-9, proficiency=1.00)]
|
|
|
|
assert rank_candidates(rows, quality_tolerance=0)[0]["model_id"] == "cheap"
|
|
|
|
|
|
# --- rejection reasons ----------------------------------------------------
|
|
#
|
|
# is_eligible returns a bool, so a dropped model used to vanish without
|
|
# explanation and "no model satisfies the hard filters" was a dead end. The
|
|
# reason string is what the debug log prints, so it has to be exact rather than
|
|
# re-derived somewhere else.
|
|
|
|
def _reason(row, **overrides):
|
|
kwargs = {
|
|
"required_context_tokens": 10_000,
|
|
"required_tier": 2,
|
|
"latency_tolerance": "interactive",
|
|
"allowed_access_levels": ["public"],
|
|
"exclude_stale": True,
|
|
"exclude_deprecated": True,
|
|
}
|
|
kwargs.update(overrides)
|
|
return rejection_reason(row, **kwargs)
|
|
|
|
|
|
def test_an_eligible_row_has_no_reason():
|
|
assert _reason(_row()) is None
|
|
|
|
|
|
def test_the_reason_names_the_filter_and_its_numbers():
|
|
assert _reason(_row(effective_context_window=5_000)) == "context(5000<10000)"
|
|
assert _reason(_row(tier=1)) == "tier(1<2)"
|
|
assert _reason(_row(availability="stale")) == "stale"
|
|
assert _reason(_row(availability="deprecated")) == "deprecated"
|
|
assert _reason(_row(access_level="canary")) == "access_level(canary)"
|
|
assert _reason(_row(latency_class="flex")) == "latency_class(flex)"
|
|
|
|
|
|
def test_unknown_values_say_unknown_rather_than_comparing():
|
|
assert _reason(_row(effective_context_window=None)) == "context(unknown)"
|
|
assert _reason(_row(tier=None)) == "tier(unknown)"
|
|
|
|
|
|
def test_the_tool_filter_reports_the_measured_score():
|
|
reason = _reason(_row(tool_proficiency=0.33), min_tool_proficiency=0.5)
|
|
|
|
assert reason == "tool_proficiency(0.33<0.5)"
|
|
|
|
|
|
def test_reasons_are_single_tokens():
|
|
"""They go straight into a logfmt value; a space would force quoting."""
|
|
for row in (_row(tier=1), _row(availability="stale"),
|
|
_row(access_level="canary"), _row(latency_class="flex"),
|
|
_row(supports_vision=0), _row(supports_vision=None),
|
|
_row(supports_json_mode=0), _row(supports_json_mode=None)):
|
|
reason = _reason(row, require_vision=True, require_json_mode=True)
|
|
assert reason is not None
|
|
assert " " not in reason
|
|
|
|
|
|
# --- capability_gate_reason (the extracted flag rule) -----------------------
|
|
#
|
|
# rejection_reason delegates its vision/json-mode arm to this function, and
|
|
# dispatcher._check_pinned_capabilities reuses it so a pinned-model check and
|
|
# routed traffic cannot drift apart. It is unit-tested directly because the
|
|
# pinned-model path has no other cheap way to exercise these branches without
|
|
# a full request round-trip.
|
|
|
|
def test_capability_gate_passes_when_not_required():
|
|
# With no capability required, a model that lacks both still passes.
|
|
assert capability_gate_reason(_row(supports_vision=0, supports_json_mode=0)) is None
|
|
|
|
|
|
def test_capability_gate_rejects_missing_vision_flag():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=None), require_vision=True
|
|
) == "vision(unknown)"
|
|
|
|
|
|
def test_capability_gate_rejects_vision_unsupported():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=0), require_vision=True
|
|
) == "vision(unsupported)"
|
|
|
|
|
|
def test_capability_gate_rejects_missing_json_mode_flag():
|
|
assert capability_gate_reason(
|
|
_row(supports_json_mode=None), require_json_mode=True
|
|
) == "json_mode(unknown)"
|
|
|
|
|
|
def test_capability_gate_rejects_json_mode_unsupported():
|
|
assert capability_gate_reason(
|
|
_row(supports_json_mode=0), require_json_mode=True
|
|
) == "json_mode(unsupported)"
|
|
|
|
|
|
def test_capability_gate_admits_a_row_that_supports_both():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=1, supports_json_mode=1),
|
|
require_vision=True, require_json_mode=True,
|
|
) is None
|
|
|
|
|
|
def test_capability_gate_unknown_reason_on_empty_row():
|
|
assert capability_gate_reason({}, require_vision=True) == "vision(unknown)"
|
|
assert capability_gate_reason({}, require_json_mode=True) == "json_mode(unknown)"
|
|
|
|
|
|
def test_is_eligible_still_agrees_with_the_reason():
|
|
"""One copy of the rules, two views of it."""
|
|
for row in (_row(), _row(tier=1), _row(latency_class="flex")):
|
|
assert is_eligible(row, **{
|
|
"required_context_tokens": 10_000, "required_tier": 2,
|
|
"latency_tolerance": "interactive", "allowed_access_levels": ["public"],
|
|
"exclude_stale": True, "exclude_deprecated": True,
|
|
}) == (_reason(row) is None)
|