"""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 config import FlexPreference from routing import ( apply_flex_preference, capability_gate_reason, credit_attenuation_multiplier, estimated_cost, get_flex_sibling, 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 # --- circuit breaker (exclude_models) ------------------------------------- def test_circuit_open_row_is_rejected_by_exclude_models(): assert _eligible(_row(model_id="dead-model"), exclude_models={"dead-model"}) is False def test_circuit_open_row_is_kept_when_not_excluded(): assert _eligible(_row(model_id="dead-model"), exclude_models=set()) is True def test_exclude_models_reason_is_circuit_open(): row = _row(model_id="dead-model") reason = rejection_reason( row, required_context_tokens=10_000, required_tier=2, latency_tolerance="interactive", allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True, exclude_models={"dead-model"}, ) assert reason == "circuit_open" # --- 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 # --- credit attenuation multiplier ---------------------------------------- def test_credit_attenuation_unknown_balance_is_healthy(): assert credit_attenuation_multiplier( None, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == 1.0 def test_credit_attenuation_above_soft_floor_is_healthy(): assert credit_attenuation_multiplier( 10.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == 1.0 def test_credit_attenuation_at_soft_floor_is_healthy(): assert credit_attenuation_multiplier( 5.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == 1.0 def test_credit_attenuation_at_zero_floor_is_max(): assert credit_attenuation_multiplier( 0.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == 5.0 def test_credit_attenuation_below_zero_floor_is_max(): assert credit_attenuation_multiplier( -1.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == 5.0 def test_credit_attenuation_midpoint_linear(): # balance 2.5 is halfway between soft=5 and zero=0 with max=5. # frac = (5 - 2.5) / (5 - 0) = 0.5 # result = 1.0 + 0.5 * (5 - 1.0) = 3.0 assert credit_attenuation_multiplier( 2.5, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0 ) == pytest.approx(3.0) def test_credit_attenuation_multiplier_range_independence(): # Same fractional position with max_multiplier=2.0 should yield 1.5. assert credit_attenuation_multiplier( 2.5, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=2.0 ) == pytest.approx(1.5) def test_credit_attenuation_equal_floors_is_safe(): # With soft == zero, the guard order (>= soft first, <= zero second) # means the interpolation branch is unreachable regardless of balance. # This construction-safety test hits the <= zero branch. assert credit_attenuation_multiplier( 0.0, soft_floor_usd=5.0, zero_floor_usd=5.0, max_multiplier=5.0 ) == 5.0 # --- 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) # --- get_flex_sibling ------------------------------------------------------- # # The flex-preference knob can route a request to a `-flex` row (deferred # during peak). For an operator who prefers flex latency but not a hard # failure, the standard winner has a flex sibling that carries identical # weights/serving class except latency_class. This helper finds it; it is # deliberately a pure function of the rows, so the dispatcher owns only the # "which preference applies" decision and not the catalog lookup. def test_flex_sibling_when_standard_row_has_one(): # Given: the standard kimi-k3 winner and its flex sibling, differing only # by latency_class selected = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard") flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex") sibling = get_flex_sibling(selected, [selected, flex]) assert sibling is not None assert sibling["model_id"] == "kimi-k3-flex" assert sibling["latency_class"] == "flex" def test_flex_sibling_must_match_all_three_dims(): # A flex row that shares base_model_id but not every serving dimension is # NOT a sibling: -fast/-short change weights or context budget. selected = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard") wrong_reasoning = _row(model_id="kimi-k3-fast-flex", base_model_id="kimi-k3", reasoning_mode="reduced", context_variant="full", latency_class="flex") wrong_ctx = _row(model_id="kimi-k3-short-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="short", latency_class="flex") # A matching flex must still be found among non-matching rows. right = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex") assert get_flex_sibling(selected, [wrong_reasoning, wrong_ctx, right])["model_id"] == "kimi-k3-flex" def test_flex_sibling_is_none_when_selected_is_already_flex(): selected = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex") assert get_flex_sibling(selected, [selected]) is None def test_flex_sibling_is_none_when_no_matching_row_exists(): # The flex row present belongs to a different base model entirely selected = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard") other = _row(model_id="glm-5.2-flex", base_model_id="glm-5.2", reasoning_mode="default", context_variant="full", latency_class="flex") assert get_flex_sibling(selected, [selected, other]) is None def test_flex_sibling_ignores_stale_or_unroutable_flex_rows(): # The helper is purely structural: it matches by identity dims regardless # of freshness/access. Routing-level gating is the caller's job. selected = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard") flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex", availability="stale") sibling = get_flex_sibling(selected, [selected, flex]) assert sibling is not None assert sibling["model_id"] == "kimi-k3-flex" # --- apply_flex_preference: the post-rank serving-class switch -------------- # # rank_candidates already decides quality and cost; this runs AFTER it to # override only the serving class when the operator's flex-preference stance # asks for it. The mode matrix below pins the behavior of every preference # under both latency tolerances, plus the no-sibling fallback. def _std_flex_pair(**overrides): """A ranked standard winner plus its weight-equivalent flex twin. The flex twin shares base_model_id/reasoning_mode/context_variant and differs only by latency_class. ``base_model_id`` defaults to the model_id (matching the get_flex_sibling tests which must set it explicitly). """ std = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard", proficiency=0.5, cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, **overrides) flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex", proficiency=0.5, cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, **overrides) return std, flex def test_apply_no_flex_never_swaps_under_interactive(): # no-flex: never route to a flex row, even when one exists and would help. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.no_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_no_flex_leaves_an_already_flex_winner_alone(): # If the rank winner is already flex (batch admitted it), no-flex does not # bounce it back to standard; it leaves the decision as ranked. flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex", proficiency=0.5) row, swapped, forced, cost = apply_flex_preference( flex, [flex], FlexPreference.no_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3-flex" assert swapped is False assert forced is False assert cost == flex["cost"] def test_apply_auto_makes_no_swap_even_when_a_flex_sibling_exists(): # auto defers to the latency filter that already ran; a flex sibling # existing does not change the ranked winner. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.auto, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_prefer_flex_holds_back_under_interactive(): # prefer-flex swaps only when the flex sibling is eligible under the # current hard filters; interactive's latency filter dismisses it. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.prefer_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_prefer_flex_swaps_under_batch(): # Under batch the latency filter admits flex, so the sibling is eligible # and the swap happens without bypassing anything. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.prefer_flex, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3-flex" assert swapped is True assert forced is False assert cost is not None def test_apply_force_flex_swaps_under_interactive_and_marks_forced(): # force-flex bypasses the interactive latency filter: swap happens and the # bypass is recorded as flex_forced=True. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3-flex" assert swapped is True assert forced is True assert cost is not None def test_apply_force_flex_swaps_under_batch_without_forced_flag(): # Under batch there is no filter to bypass, so force-flex swaps but the # forced flag stays False. std, flex = _std_flex_pair() row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3-flex" assert swapped is True assert forced is False assert cost is not None def test_apply_falls_back_when_no_flex_sibling_exists(): # No flex twin in the catalog: every preference falls back to the ranked # winner and its cost. std = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard", proficiency=0.5) for pref in (FlexPreference.no_flex, FlexPreference.auto, FlexPreference.prefer_flex, FlexPreference.force_flex): row, swapped, forced, cost = apply_flex_preference( std, [std], pref, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_reestimates_cost_for_the_swapped_flex_row(): # The flex twin carries the same list price as its standard sibling, so a # swap recomputes the estimate from the flex row's own pricing fields. Give # the flex row a distinct price and confirm the returned cost tracks it. std = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard", proficiency=0.5, cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, cost_per_1m_prompt_cached=0.5) flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex", proficiency=0.5, cost_per_1m_prompt=2.0, cost_per_1m_completion=4.0, cost_per_1m_prompt_cached=1.0) row, swapped, _, _ = apply_flex_preference( std, [std, flex], FlexPreference.prefer_flex, "batch", prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert swapped is True # estimated_cost scales prompt/completion prices to this request shape, so # the flex row's higher price must drive a higher estimate than standard's. std_cost = rank_candidates( [std], prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5 )[0]["cost"] flex_cost = rank_candidates( [flex], prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5 )[0]["cost"] assert row["cost"] == flex_cost assert row["cost"] > std_cost def test_apply_prefer_flex_refuses_stale_sibling_under_batch(): # A stale flex sibling fails the freshness filter; prefer-flex under batch # must keep the ranked standard winner rather than dispatch to staleness. std, flex = _std_flex_pair(availability="stale") row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.prefer_flex, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_force_flex_refuses_stale_sibling_under_interactive(): # force-flex bypasses the latency filter, but it still must not dispatch to # a stale sibling; the standard winner stands. std, flex = _std_flex_pair(availability="stale") row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_prefer_flex_refuses_canary_sibling_under_batch(): # An access-restricted flex sibling must not be swapped to when the caller # is restricted to public rows, even under batch where latency would admit it. std, flex = _std_flex_pair(access_level="canary") row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.prefer_flex, "batch", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_force_flex_refuses_canary_sibling_under_interactive(): # force-flex bypasses only the latency filter — access restriction still # gates the sibling, so the standard winner stands. std, flex = _std_flex_pair(access_level="canary") row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "interactive", prompt_tokens=1000, completion_tokens=500, cache_rate=0.5, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert row["model_id"] == "kimi-k3" assert swapped is False assert forced is False assert cost == std["cost"] def test_apply_force_flex_prices_swapped_row_at_challenger_cache_rate(): # When challenger_cache_rate is set, the flex twin is priced at that rate # rather than the caller's cache_rate — the twin is billed as a cold # challenger because the session's incumbent cache does not carry across # the swap, keeping parity with the ranking dial at every setting. std = _row(model_id="kimi-k3", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="standard", proficiency=0.5, cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, cost_per_1m_prompt_cached=0.5) flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3", reasoning_mode="default", context_variant="full", latency_class="flex", proficiency=0.5, cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, cost_per_1m_prompt_cached=0.5) cache_rate = 0.8 challenger_rate = 0.1 row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "interactive", prompt_tokens=10_000, completion_tokens=500, cache_rate=cache_rate, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True, challenger_cache_rate=challenger_rate) assert row["model_id"] == "kimi-k3-flex" assert swapped is True assert forced is True # The cost must track the challenger rate, not the standard cached rate. assert cost == estimated_cost(flex, 10_000, 500, challenger_rate) assert cost != estimated_cost(flex, 10_000, 500, cache_rate) def test_apply_challenger_cache_rate_none_falls_back_to_cache_rate(): # Without challenger_cache_rate (the common caller case), swap behavior is # unchanged — the flex twin prices at the same cache_rate as before. std, flex = _std_flex_pair() cache_rate = 0.5 row, swapped, forced, cost = apply_flex_preference( std, [std, flex], FlexPreference.force_flex, "interactive", prompt_tokens=10_000, completion_tokens=500, cache_rate=cache_rate, required_context_tokens=1000, required_tier=2, allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True) assert swapped is True # Default None → use cache_rate, so the cost must match what we get when # we explicitly pass cache_rate. assert cost == estimated_cost(flex, 10_000, 500, cache_rate) # --- eligible_categories hard filter ---------------------------------------- def test_category_restricted_row_rejected_when_task_outside_list(): # A row with eligible_categories=["coding"] must be ineligible for a # summarization task. row = _row(eligible_categories=["coding", "refactoring"]) assert _eligible(row, task_category="summarization") is False def test_category_restricted_row_eligible_when_task_in_list(): # The same row is eligible when the task category is in the list. row = _row(eligible_categories=["coding", "refactoring"]) assert _eligible(row, task_category="refactoring") is True def test_null_eligible_categories_unaffected_by_task(): # NULL eligible_categories NEVER rejects — not even when task_category is # provided, and not when it is None. row = _row() # missing eligible_categories, behaves as None assert _eligible(row, task_category="coding") is True assert _eligible(row, task_category=None) is True def test_raw_comma_string_is_parsed_not_substring_matched(): """SQLite stores this column as a comma-joined string, and `in` on a string is a SUBSTRING test — so an unparsed row admitted `summarization` to a model eligible only for `file_summarization`. dispatcher parsed the column; admin.py's profile probe passed rows straight out of SELECT *. """ row = _row(eligible_categories="file_summarization,diff_checking") assert _eligible(row, task_category="summarization") is False assert _eligible(row, task_category="file_summarization") is True assert _eligible(row, task_category="diff_checking") is True def test_parse_eligible_categories_is_idempotent(): """Called on rows that may or may not have been normalized already.""" from routing import parse_eligible_categories assert parse_eligible_categories(None) is None once = parse_eligible_categories(" a , b ,, ") assert once == ["a", "b"] assert parse_eligible_categories(once) == ["a", "b"] def test_category_ineligible_reason_is_single_token(): # Reason strings go into logfmt values — no spaces allowed. row = _row(eligible_categories=["coding"]) reason = _reason(row, task_category="summarization") assert reason == "category_ineligible" assert " " not in reason # --- restrict_to profile allowlist ---------------------------------------- def test_restrict_to_none_is_unrestricted(): row = _row(model_id="m1") assert _eligible(row, restrict_to=None) is True def test_restrict_to_allows_model_in_set(): row = _row(model_id="m1") assert _eligible(row, restrict_to={"m1"}) is True def test_restrict_to_excludes_model_not_in_set(): row = _row(model_id="m2") assert _eligible(row, restrict_to={"m1"}) is False def test_restrict_to_empty_set_excludes_every_model(): row = _row(model_id="m1") assert _eligible(row, restrict_to=set()) is False def test_restrict_to_reason_is_profile_excluded(): row = _row(model_id="m2") reason = _reason(row, restrict_to={"m1"}) assert reason == "profile_excluded" def test_restrict_to_is_single_token_reason(): row = _row(model_id="m2") reason = _reason(row, restrict_to={"m1"}) assert " " not in reason def test_select_candidates_applies_restrict_to_allowlist(): rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")] 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, restrict_to={"m1"}) assert [r["model_id"] for r in selected] == ["m1"] def test_select_candidates_empty_restrict_to_yields_no_candidates(): rows = [_row(model_id="m1"), _row(model_id="m2")] 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, restrict_to=set()) assert selected == [] def test_restrict_to_and_exclude_models_compose_as_and(): # An allowlist of {m1, m2} plus a circuit-open exclusion of m1 should # leave only m2. rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")] 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, exclude_models={"m1"}, restrict_to={"m1", "m2"}) assert [r["model_id"] for r in selected] == ["m2"] def test_select_candidates_drops_outside_category(): # Even a perfect row (tier 1, zero cost) is dropped for the wrong category. rows = [ _row( model_id="restricted", tier=1, cost=0.01, proficiency=1.0, eligible_categories=["coding"], ), _row( model_id="open", tier=2, cost=1.0, proficiency=0.5, ), ] filtered = select_candidates( rows, required_context_tokens=1000, required_tier=1, latency_tolerance="interactive", allowed_access_levels=["public"], exclude_stale=True, exclude_deprecated=True, task_category="summarization") ids = {r["model_id"] for r in filtered} assert "restricted" not in ids assert "open" in ids def test_through_is_eligible_via_filters(): # task_category must reach rejection_reason through the **filters passthrough # in is_eligible, not only via direct parameter passing. row = _row(eligible_categories=["coding"]) 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, task_category="summarization", ) is False # QA cases (a)-(g) for provider_cost_multipliers effective_cost tiebreak. def _ranked_ids(rows, **kwargs): return [r["model_id"] for r in rank_candidates(rows, **kwargs)] def test_provider_cost_multiplier_flips_tiebreak_toward_healthy_provider(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.80), _row(model_id="B", provider="provB", cost=0.10, proficiency=0.80), ] assert _ranked_ids(rows) == ["A", "B"] assert _ranked_ids( rows, provider_cost_multipliers={"provA": 5.0} ) == ["B", "A"] def test_provider_cost_multiplier_cannot_override_quality_band_gap(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.95), _row(model_id="B", provider="provB", cost=0.01, proficiency=0.80), ] assert _ranked_ids( rows, provider_cost_multipliers={"provA": 5.0}, quality_tolerance=0.1 ) == ["A", "B"] def test_ranked_rows_carry_raw_cost_and_effective_cost(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.80), ] ranked = rank_candidates(rows, provider_cost_multipliers={"provA": 5.0}) assert len(ranked) == 1 assert ranked[0]["cost"] == pytest.approx(0.10) assert ranked[0]["cost_score"] == pytest.approx(1.0) assert ranked[0]["effective_cost"] == pytest.approx(0.50) def test_provider_multiplier_applies_to_fallback_cost_when_estimate_missing(): rows = [ _row( model_id="A", provider="provA", cost=0.20, proficiency=0.80, prompt_cost_1m=None, completion_cost_1m=None, ), ] ranked = rank_candidates( rows, prompt_tokens=0, completion_tokens=0, provider_cost_multipliers={"provA": 3.0}, ) assert ranked[0]["cost"] == pytest.approx(0.20) assert ranked[0]["effective_cost"] == pytest.approx(0.60) def test_provider_cost_multiplier_unknown_provider_ignored(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.80), ] ranked = rank_candidates( rows, provider_cost_multipliers={"provUnknown": 99.0} ) assert ranked[0]["effective_cost"] == pytest.approx(0.10) def test_provider_cost_multiplier_defaults_to_one_for_unmapped_provider(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.80), _row(model_id="B", provider="provB", cost=0.10, proficiency=0.80), ] assert _ranked_ids( rows, provider_cost_multipliers={"provA": 5.0} ) == ["B", "A"] def test_provider_cost_multiplier_empty_dict_matches_no_argument(): rows = [ _row(model_id="A", provider="provA", cost=0.10, proficiency=0.80), _row(model_id="B", provider="provB", cost=0.10, proficiency=0.80), ] assert _ranked_ids(rows) == _ranked_ids( rows, provider_cost_multipliers={} ) # --- incumbency pricing: per-candidate cache pricing in rank_candidates ----- class TestIncumbentPricing: """8 tests (a)-(h) for per-candidate cache pricing inside rank_candidates. Uses cached-price fields (cost_per_1m_prompt_cached != cost_per_1m_prompt) so that varying cache_rate produces differentiable costs. """ INCUMBENT = ("provA", "incumbent-model") INCUMBENT_PROVIDER, INCUMBENT_MODEL = INCUMBENT # Consistent across all tests that need estimated_cost: _PTOKENS = 1000 _CTOKENS = 500 _PROMPT_PRICE = 10.0 _CACHED_PRICE = 1.0 _COMPLETION_PRICE = 2.0 @classmethod def _pricing_rows(cls): """Two rows with cached pricing and same-band proficiency.""" return [ _row( model_id=cls.INCUMBENT_MODEL, provider=cls.INCUMBENT_PROVIDER, proficiency=0.90, cost_per_1m_prompt=cls._PROMPT_PRICE, cost_per_1m_prompt_cached=cls._CACHED_PRICE, cost_per_1m_completion=cls._COMPLETION_PRICE, ), _row( model_id="challenger", provider="provB", proficiency=0.90, cost_per_1m_prompt=cls._PROMPT_PRICE, cost_per_1m_prompt_cached=cls._CACHED_PRICE, cost_per_1m_completion=cls._COMPLETION_PRICE, ), ] # (a) EQUIVALENCE: incumbent=None produces pre-change ranking def test_equivalence_incumbent_none_matches_default(self): rows = [ _row(model_id="A", proficiency=0.95, cost=0.10), _row(model_id="B", proficiency=0.80, cost=0.01), ] default = rank_candidates(rows) explicit = rank_candidates(rows, incumbent=None) assert [r["model_id"] for r in default] == [r["model_id"] for r in explicit] for d, e in zip(default, explicit): assert d["cost"] == e["cost"] assert d["effective_cost"] == e["effective_cost"] assert d["proficiency_score"] == pytest.approx(e["proficiency_score"]) # (b) Incumbent protection: same band + dial OFF neutral → incumbent wins def test_incumbent_wins_on_cost_when_measured_rate_is_lower(self): rows = self._pricing_rows() # cache_rate=0.5 → cheaper (more tokens at $1 cached, fewer at $10 fresh) # challenger_cache_rate=0.1 → clamp=min(0.1, 0.5)=0.1 → more expensive ranked = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: 0.5}, challenger_cache_rate=0.1, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, ) assert ranked[0]["model_id"] == self.INCUMBENT_MODEL assert ranked[0]["cost"] < ranked[1]["cost"] # (c) Quality wins outright: challenger above tolerance band beats incumbent def test_quality_gap_still_decides_outright(self): rows = [ _row( model_id=self.INCUMBENT_MODEL, provider=self.INCUMBENT_PROVIDER, proficiency=0.80, cost_per_1m_prompt=self._PROMPT_PRICE, cost_per_1m_prompt_cached=self._CACHED_PRICE, cost_per_1m_completion=self._COMPLETION_PRICE, ), _row( model_id="superior-challenger", provider="provB", proficiency=1.00, cost_per_1m_prompt=100.0, cost_per_1m_prompt_cached=10.0, cost_per_1m_completion=20.0, ), ] ranked = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: 0.5}, challenger_cache_rate=0.1, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, quality_tolerance=0.1, ) assert ranked[0]["model_id"] == "superior-challenger" # (d) Challenger cost matches estimated_cost with min(dial, inc_rate) at 3 dials def test_challenger_cost_matches_clamped_rate(self): rows = self._pricing_rows() inc_rate = 0.7 cache_rate = 0.84 for challenger_cache_rate, expected_rate in [ (None, cache_rate), # neutral → unchanged (cache_rate, cache_rate), # neutral → unchanged (0.5, min(0.5, inc_rate)), # mid → clamped (0.0, min(0.0, inc_rate)), # 0.0 → clamped ]: ranked = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: inc_rate}, challenger_cache_rate=challenger_cache_rate, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, cache_rate=cache_rate, ) # Challenger is always index 1 challenger = ranked[1] expected = estimated_cost( rows[1], self._PTOKENS, self._CTOKENS, expected_rate ) assert challenger["cost"] == pytest.approx(expected) # (e) Rate ladder: dict hit → source="measured"; dict miss → source="assumed" def test_rate_ladder_source_metadata(self): rows = self._pricing_rows() # Hit: incumbent in measured_cache_rates ranked_hit = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: 0.5}, challenger_cache_rate=0.3, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, ) assert ranked_hit[0]["incumbent_pricing"]["source"] == "measured" # Miss: measured_cache_rates doesn't include incumbent rows2 = self._pricing_rows() ranked_miss = rank_candidates( rows2, incumbent=self.INCUMBENT, measured_cache_rates={}, # empty → no entry challenger_cache_rate=0.3, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, ) assert ranked_miss[0]["incumbent_pricing"]["source"] == "assumed" # (f) Composition: effective_cost == cost * provider_cost_multipliers[provider] def test_effective_cost_applies_multiplier_for_incumbent(self): rows = self._pricing_rows() ranked = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: 0.5}, challenger_cache_rate=0.3, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, provider_cost_multipliers={self.INCUMBENT_PROVIDER: 5.0}, ) # Find the incumbent row by model_id (it may not be ranked[0]) inc_row = [r for r in ranked if r["model_id"] == self.INCUMBENT_MODEL][0] assert inc_row["effective_cost"] == pytest.approx( inc_row["cost"] * 5.0 ) # (g) NEUTRAL-EQUIVALENCE BLOCKING TEST: both forms byte-identical to reference def test_neutral_forms_are_byte_identical_to_reference(self): rows = self._pricing_rows() # Pre-condition: incumbent IS in measured_cache_rates (non-vacuous) rates = {self.INCUMBENT: 0.3} assert self.INCUMBENT in rates # Reference: no incumbent pricing ref = rank_candidates(rows, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS) # Form (i): challenger_cache_rate=None form1 = rank_candidates(rows, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, incumbent=self.INCUMBENT, measured_cache_rates=rates, challenger_cache_rate=None) # Form (ii): challenger_cache_rate == cache_rate form2 = rank_candidates(rows, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, incumbent=self.INCUMBENT, measured_cache_rates=rates, challenger_cache_rate=0.84) # Explicit field-by-field comparison for byte identity for f in form1, form2: assert len(f) == len(ref) for i in range(len(ref)): for key in ref[i]: assert ref[i][key] == f[i][key], \ f"Mismatch at row {i}, key {key}: {ref[i][key]} != {f[i][key]}" # (h) INVARIANT PROPERTY TEST: incumbent_rate >= every challenger_rate def test_incumbent_rate_always_ge_challenger_rate(self): rows = self._pricing_rows() for inc_rate in (0.731, 0.953): # straddle assumed cache_rate (0.84) for dial in (0.0, 0.1, 0.3, 0.5, 0.7, 0.9, 1.0): ranked = rank_candidates( rows, incumbent=self.INCUMBENT, measured_cache_rates={self.INCUMBENT: inc_rate}, challenger_cache_rate=dial, prompt_tokens=self._PTOKENS, completion_tokens=self._CTOKENS, ) inc_row = next( r for r in ranked if r["model_id"] == self.INCUMBENT_MODEL ) chal_row = next( r for r in ranked if r["model_id"] != self.INCUMBENT_MODEL ) # THE CLAMP pinned: challenger priced at min(dial, inc_rate) # for EVERY dial — including every dial ABOVE the incumbent's # measured rate (the inversion zone). Without the clamp # (pricing challengers at the raw dial) this fails. assert chal_row["cost"] == pytest.approx(estimated_cost( next(r for r in rows if r["model_id"] == chal_row["model_id"]), self._PTOKENS, self._CTOKENS, min(dial, inc_rate), )), f"inc_rate={inc_rate} dial={dial}: clamp violated" assert inc_row["cost"] == pytest.approx(estimated_cost( next(r for r in rows if r["model_id"] == inc_row["model_id"]), self._PTOKENS, self._CTOKENS, inc_rate, )) # Invariant in cost terms: cached is cheaper than fresh, so # the challenger's rate (min(dial, inc_rate)) never exceeds # the incumbent's measured rate — it can never cost less. assert chal_row["cost"] >= inc_row["cost"]