808 lines
31 KiB
Python
808 lines
31 KiB
Python
"""Tests for session incumbent lookup and hard-filter eviction.
|
|
|
|
Covers ``_session_incumbent_lookup`` (DB-backed retrieval by session key
|
|
with ``kind`` allowlist filtering) and ``_resolve_incumbent`` (hard-filter
|
|
eviction of a retrieved incumbent tuple). Both are module-level helpers in
|
|
``dispatcher.py``, and neither depends on FastAPI or the full route path.
|
|
|
|
The database tests use the same offline temp-DB pattern as
|
|
``test_route_decisions.py``: a throwaway SQLite file seeded from
|
|
``config/schema.sql``, never the live ``router.db``.
|
|
"""
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from config import Objective
|
|
from dispatcher import _resolve_incumbent, _session_incumbent_lookup
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def db_path():
|
|
"""Provision a throwaway SQLite DB with the full schema."""
|
|
tmp = Path(tempfile.mkdtemp()) / "test_incumbent.db"
|
|
conn = sqlite3.connect(str(tmp))
|
|
conn.executescript(SCHEMA_SQL)
|
|
conn.close()
|
|
return str(tmp)
|
|
|
|
|
|
def _conn(db_path: str, row_factory=sqlite3.Row):
|
|
"""Helper to open a connection with the right row factory."""
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = row_factory
|
|
return conn
|
|
|
|
|
|
def _make_db(tmp_path):
|
|
"""Provision a throwaway SQLite DB with the full schema."""
|
|
path = str(tmp_path / "test_incumbent.db")
|
|
conn = sqlite3.connect(path)
|
|
conn.executescript(SCHEMA_SQL)
|
|
conn.close()
|
|
return path
|
|
|
|
|
|
def _route_conn(db_path):
|
|
"""Open a read-write connection for seeding route data."""
|
|
conn = sqlite3.connect(db_path)
|
|
return conn
|
|
# Helpers for seeding route_decisions rows
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ROUTE_DECISIONS_COLS = [
|
|
"observed_at", "kind", "session_key",
|
|
"selected_provider", "selected_model",
|
|
"task_category", "task_tier", "required_context_tokens",
|
|
"confidence", "classifier_ms", "classification_source",
|
|
"latency_tolerance", "candidates_considered",
|
|
]
|
|
_ROUTE_DECISIONS_COLS_STR = ", ".join(_ROUTE_DECISIONS_COLS)
|
|
|
|
|
|
def _seed_decision(conn, *,
|
|
kind="chat",
|
|
session_key="ses_test",
|
|
selected_provider="test-provider",
|
|
selected_model="test-model",
|
|
**overrides):
|
|
"""Insert a minimal route_decisions row, returning the id."""
|
|
row = {
|
|
"observed_at": "2026-09-13T00:00:00+00:00",
|
|
"kind": kind,
|
|
"session_key": session_key,
|
|
"selected_provider": selected_provider,
|
|
"selected_model": selected_model,
|
|
"task_category": "general",
|
|
"task_tier": 2,
|
|
"required_context_tokens": 4096,
|
|
"confidence": 0.95,
|
|
"classifier_ms": 42,
|
|
"classification_source": "classifier",
|
|
"latency_tolerance": "interactive",
|
|
"candidates_considered": 5,
|
|
}
|
|
row.update(overrides)
|
|
cols = ", ".join(row)
|
|
placeholders = ", ".join("?" * len(row))
|
|
conn.execute(
|
|
f"INSERT INTO route_decisions ({cols}) VALUES ({placeholders})",
|
|
tuple(row.values()),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _session_incumbent_lookup tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSessionIncumbentLookup:
|
|
"""Tests for the DB-backed incumbent retrieval."""
|
|
|
|
def test_second_turn_returns_first_turn_pair(self, db_path):
|
|
"""(a) Second turn of a chat session returns the first turn's tuple."""
|
|
conn = _conn(db_path)
|
|
_seed_decision(conn, session_key="ses_two",
|
|
selected_provider="prov-a", selected_model="model-x",
|
|
kind="chat")
|
|
_seed_decision(conn, session_key="ses_two",
|
|
selected_provider="prov-b", selected_model="model-y",
|
|
kind="chat")
|
|
result = _session_incumbent_lookup(conn, "ses_two")
|
|
# The second INSERT is chronologically later, so it should come back.
|
|
assert result == ("prov-b", "model-y")
|
|
conn.close()
|
|
|
|
def test_fresh_session_returns_none(self, db_path):
|
|
"""(b) Fresh session with no rows returns None."""
|
|
conn = _conn(db_path)
|
|
result = _session_incumbent_lookup(conn, "never_seen")
|
|
assert result is None
|
|
conn.close()
|
|
|
|
@pytest.mark.parametrize("kind,label", [
|
|
("route", "route probe"),
|
|
("passthrough", "passthrough pin"),
|
|
("local_dispatch_fallback", "local dispatch fallback"),
|
|
("dispatch", "dispatch entry"),
|
|
])
|
|
def test_non_chat_kinds_never_become_incumbent(self, db_path, kind, label):
|
|
"""(e,f) Non-chat kinds do NOT set the incumbent.
|
|
|
|
``kind='route'``, ``kind='passthrough'``,
|
|
``kind='local_dispatch_fallback'``, and ``kind='dispatch'`` are all
|
|
excluded by the allowlist ``kind IN ('chat',)``.
|
|
"""
|
|
conn = _conn(db_path)
|
|
_seed_decision(conn, session_key="ses_kind",
|
|
kind=kind,
|
|
selected_provider="prov-a",
|
|
selected_model="model-z")
|
|
# Also seed a chat row to prove it's the kind filter, not emptiness.
|
|
_seed_decision(conn, session_key="ses_kind",
|
|
kind="chat",
|
|
selected_provider="prov-b",
|
|
selected_model="model-y")
|
|
result = _session_incumbent_lookup(conn, "ses_kind")
|
|
# The chat row should be the only one eligible, not the non-chat row.
|
|
assert result == ("prov-b", "model-y"), (
|
|
f"kind={kind!r} ({label}) must NOT shadow the chat row"
|
|
)
|
|
conn.close()
|
|
|
|
def test_returns_none_when_selected_model_is_null(self, db_path):
|
|
"""A row with a null selected_model is skipped."""
|
|
conn = _conn(db_path)
|
|
_seed_decision(conn, session_key="ses_null",
|
|
selected_model=None, selected_provider=None)
|
|
assert _session_incumbent_lookup(conn, "ses_null") is None
|
|
conn.close()
|
|
|
|
def test_prefers_latest_id(self, db_path):
|
|
"""When multiple chat rows exist, the highest id wins."""
|
|
conn = _conn(db_path)
|
|
_seed_decision(conn, session_key="ses_multi",
|
|
selected_provider="old", selected_model="v1",
|
|
kind="chat")
|
|
_seed_decision(conn, session_key="ses_multi",
|
|
selected_provider="new", selected_model="v2",
|
|
kind="chat")
|
|
assert _session_incumbent_lookup(conn, "ses_multi") == ("new", "v2")
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_incumbent tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def sample_rows():
|
|
"""A small in-memory catalog that mimics the rows passed to route()."""
|
|
return [
|
|
{"model_id": "model-a", "provider": "prov-a",
|
|
"effective_context_window": 32_000, "tier": 2,
|
|
"latency_class": "standard", "access": "public",
|
|
"availability": "available", "deprecated": False, "stale": False},
|
|
{"model_id": "model-b", "provider": "prov-b",
|
|
"effective_context_window": 128_000, "tier": 3,
|
|
"latency_class": "standard", "access": "public",
|
|
"availability": "available", "deprecated": False, "stale": False},
|
|
{"model_id": "model-c", "provider": "prov-a",
|
|
"effective_context_window": 8_000, "tier": 1,
|
|
"latency_class": "batch", "access": "private",
|
|
"availability": "available", "deprecated": False, "stale": False},
|
|
]
|
|
|
|
|
|
def _base_filters():
|
|
"""Default filters that pass every row above."""
|
|
return dict(
|
|
required_context_tokens=4_096,
|
|
required_tier=1,
|
|
latency_tolerance="interactive",
|
|
allowed_access_levels=("public", "private"),
|
|
exclude_stale=False,
|
|
exclude_deprecated=False,
|
|
exclude_models=set(),
|
|
min_tool_proficiency=None,
|
|
require_vision=False,
|
|
require_json_mode=False,
|
|
task_category=None,
|
|
restrict_to=None,
|
|
)
|
|
|
|
|
|
class TestResolveIncumbent:
|
|
"""Tests for hard-filter eviction of the incumbent."""
|
|
|
|
def test_incumbent_survives_default_filters(self, sample_rows):
|
|
"""A valid incumbent passes through unchanged."""
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to=None,
|
|
filters=_base_filters(),
|
|
)
|
|
assert result == ("prov-a", "model-a")
|
|
|
|
def test_none_incumbent_returns_none(self, sample_rows):
|
|
"""No incumbent → no result."""
|
|
result = _resolve_incumbent(
|
|
sample_rows, None, restrict_to=None, filters=_base_filters(),
|
|
)
|
|
assert result is None
|
|
|
|
def test_incumbent_not_in_rows_returns_none(self, sample_rows):
|
|
"""Incumbent model absent from catalog → None."""
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-unknown", "model-ghost"),
|
|
restrict_to=None,
|
|
filters=_base_filters(),
|
|
)
|
|
assert result is None
|
|
|
|
def test_evicted_by_restrict_to(self, sample_rows):
|
|
"""(c) Profile switch excluding the incumbent evicts via restrict_to."""
|
|
# restrict_to only admits model-b, but incumbent is model-a
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to={"model-b"},
|
|
filters=_base_filters(),
|
|
)
|
|
assert result is None
|
|
|
|
def test_not_evicted_when_restrict_to_includes(self, sample_rows):
|
|
"""Incumbent survives when restrict_to includes it."""
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to={"model-a", "model-b"},
|
|
filters=_base_filters(),
|
|
)
|
|
assert result == ("prov-a", "model-a")
|
|
|
|
def test_evicted_by_context_ceiling(self, sample_rows):
|
|
"""(d) Incumbent exceeding context ceiling evicts via rejection_reason.
|
|
|
|
model-a has effective_context_window=32_000. Setting
|
|
required_context_tokens above that should evict.
|
|
"""
|
|
filters = _base_filters()
|
|
filters["required_context_tokens"] = 64_000
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to=None,
|
|
filters=filters,
|
|
)
|
|
assert result is None
|
|
|
|
def test_evicted_by_tier_filter(self, sample_rows):
|
|
"""Incumbent below required tier evicts."""
|
|
# model-a is tier 2; require tier 3
|
|
filters = _base_filters()
|
|
filters["required_tier"] = 3
|
|
result = _resolve_incumbent(
|
|
sample_rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to=None,
|
|
filters=filters,
|
|
)
|
|
assert result is None
|
|
|
|
def test_evicted_by_latency_mismatch(self, sample_rows):
|
|
"""Incumbent with 'flex' latency class evicted in interactive mode."""
|
|
rows = [
|
|
{**sample_rows[0], "latency_class": "flex"},
|
|
sample_rows[1],
|
|
sample_rows[2],
|
|
]
|
|
filters = _base_filters()
|
|
filters["latency_tolerance"] = "interactive"
|
|
result = _resolve_incumbent(
|
|
rows,
|
|
("prov-a", "model-a"),
|
|
restrict_to=None,
|
|
filters=filters,
|
|
)
|
|
assert result is None
|
|
|
|
def test_evicted_by_stale(self, sample_rows):
|
|
"""Incumbent with stale availability evicts."""
|
|
rows = [
|
|
{**sample_rows[0], "availability": "stale", "stale": True},
|
|
]
|
|
filters = _base_filters()
|
|
filters["exclude_stale"] = True
|
|
result = _resolve_incumbent(
|
|
rows, ("prov-a", "model-a"), restrict_to=None, filters=filters,
|
|
)
|
|
assert result is None
|
|
|
|
def test_requires_full_row(self, sample_rows):
|
|
"""A row with no effective_context_window (None) is rejected."""
|
|
rows = [
|
|
{**sample_rows[0], "effective_context_window": None},
|
|
]
|
|
result = _resolve_incumbent(
|
|
rows, ("prov-a", "model-a"), restrict_to=None,
|
|
filters=_base_filters(),
|
|
)
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Call-site integration tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIncumbentFlagGate:
|
|
"""Tests for the route() call-site stub.
|
|
|
|
These verify that the incumbent-lookup block only runs when the feature
|
|
flag is on, without needing the full FastAPI test client.
|
|
"""
|
|
|
|
def test_incumbent_unset_when_flag_off(self, tmp_path, monkeypatch):
|
|
"""(g) Flag off → no DB query issued, no incumbent trio constructed."""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
conn.close()
|
|
|
|
spy_lookup = MagicMock(side_effect=dispatcher._session_incumbent_lookup)
|
|
monkeypatch.setattr(dispatcher, "_session_incumbent_lookup", spy_lookup)
|
|
spy_rates = MagicMock(side_effect=dispatcher._measured_cache_rates)
|
|
monkeypatch.setattr(dispatcher, "_measured_cache_rates", spy_rates)
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", False)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
result = _route_via_client(client)
|
|
assert result["selected"] is not None
|
|
|
|
# Neither the lookup nor the measured-rates path fires when flag is off.
|
|
assert spy_lookup.call_count == 0, "_session_incumbent_lookup called with flag off"
|
|
assert spy_rates.call_count == 0, "_measured_cache_rates called with flag off"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T6 call-site wiring tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
import logging
|
|
|
|
from starlette.testclient import TestClient
|
|
|
|
import dispatcher
|
|
|
|
CACHE_RATE_COLS = (
|
|
"observed_at", "kind", "session_key",
|
|
"selected_provider", "selected_model",
|
|
"task_category", "task_tier", "required_context_tokens",
|
|
"confidence", "classifier_ms", "classification_source",
|
|
"latency_tolerance", "candidates_considered",
|
|
"provider", "model_id",
|
|
"base_cost_per_1m_prompt", "base_cost_per_1m_completion",
|
|
"prompt_tokens", "cached_prompt_tokens"
|
|
)
|
|
CACHE_RATE_COLS_STR = ", ".join(CACHE_RATE_COLS)
|
|
|
|
|
|
def _seed_cache_data(conn, provider, model_id, prompt_tokens, cached_tokens,
|
|
observed_at=None):
|
|
"""Insert a minimal energy_observations row for cache_rate_series."""
|
|
if observed_at is None:
|
|
from datetime import datetime, timezone
|
|
observed_at = datetime.now(timezone.utc).isoformat()
|
|
conn.execute(
|
|
f"INSERT INTO energy_observations ({CACHE_RATE_COLS_STR}) VALUES ({', '.join('?' * len(CACHE_RATE_COLS))})",
|
|
(
|
|
observed_at, "dispatch", None,
|
|
None, None, # session_key, provider, model in old schema
|
|
"general", 2, 4096,
|
|
0.95, 42, "classifier",
|
|
"interactive", 3,
|
|
provider, model_id,
|
|
0.50, 0.50,
|
|
prompt_tokens, cached_tokens,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _seed_models_for_route(conn):
|
|
"""Two providers with identical profiling so routing goes by raw cost."""
|
|
conn.executemany(
|
|
"""INSERT INTO models (
|
|
model_id, provider, base_model_id, tier, context_window,
|
|
effective_context_window, max_output_tokens,
|
|
cost_per_1m_prompt, cost_per_1m_completion,
|
|
supports_vision, supports_json_mode,
|
|
latency_class, reasoning_mode, context_variant,
|
|
access_level, availability, last_updated
|
|
) VALUES (?, ?, 'baseline', ?, ?, ?, ?, ?, ?,
|
|
1, 1, 'standard', 'default', 'full', 'public', 'active', '2026-09-01T00:00:00+00:00')
|
|
""",
|
|
[
|
|
("nw-model", "nw", 2, 131072, 131072, 4096, 0.50, 0.50),
|
|
("or-model", "or", 2, 131072, 131072, 4096, 0.50, 0.50),
|
|
],
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _seed_incumbent_session(conn):
|
|
"""Seed a route_decisions row so _session_incumbent_lookup returns an incumbent."""
|
|
conn.execute(
|
|
f"INSERT INTO route_decisions ({_ROUTE_DECISIONS_COLS_STR}) VALUES ({', '.join('?' * len(_ROUTE_DECISIONS_COLS))})",
|
|
(
|
|
"2026-09-13T00:00:00+00:00", "chat", "ses_w6",
|
|
"nw", "nw-model",
|
|
"general", 2, 4096, 0.95, 42, "classifier",
|
|
"interactive", 3,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _seed_measured_cache_rate(conn, provider, model_id,
|
|
prompt_tokens=1000, cached_tokens=800,
|
|
observations=30):
|
|
"""Seed energy_observations so _measured_cache_rates returns a rate.
|
|
|
|
objective.incumbent_rate_min_observations defaults to 25, so we seed
|
|
enough rows to clear the floor.
|
|
"""
|
|
for _ in range(observations):
|
|
conn.execute(
|
|
"""INSERT INTO energy_observations (
|
|
model_id, provider, task_category,
|
|
prompt_tokens, completion_tokens, cached_prompt_tokens,
|
|
cached_tokens_source, observed_at
|
|
) VALUES (?, ?, 'general', ?, 500, ?, 'reported',
|
|
'2026-09-13T00:00:00+00:00')""",
|
|
(model_id, provider, prompt_tokens, cached_tokens),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _route_via_client(client):
|
|
"""POST /route with a task-category override so no classifier is needed."""
|
|
resp = client.post(
|
|
"/route",
|
|
json={
|
|
"task": "Fix a bug",
|
|
"task_category": "general",
|
|
"task_tier": 2,
|
|
"required_context_tokens": 4096,
|
|
},
|
|
)
|
|
return resp.json()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_measured_rate_cache():
|
|
"""Each test starts with a cold measured-rates cache and no session key.
|
|
|
|
The ContextVar set inside T6 tests would otherwise leak into other test
|
|
modules in the same run (a ContextVar set in this module's context
|
|
persists for every later test in the same pytest process).
|
|
"""
|
|
dispatcher._measured_rate_cache = None
|
|
dispatcher._measured_rate_cached_at = 0.0
|
|
token = dispatcher._current_session_key.set(None)
|
|
yield
|
|
dispatcher._measured_rate_cache = None
|
|
dispatcher._measured_rate_cached_at = 0.0
|
|
dispatcher._current_session_key.reset(token)
|
|
|
|
|
|
# --- (a) Flag off → no incumbency wiring ---
|
|
|
|
class TestFlagOff:
|
|
def test_no_incumbent_kwargs_when_flag_off(self, tmp_path, monkeypatch):
|
|
"""(a) Flag off → rank_candidates called without incumbent kwargs.
|
|
|
|
Even with session history (incumbent row in route_decisions), the
|
|
incumbency block at route() is entirely skipped, so the fast path
|
|
produces byte-identical results to the pre-change call.
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", False)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
result = _route_via_client(client)
|
|
# Baseline behavior: selects the first catalog row by cost
|
|
assert result["selected"]["provider"] == "nw"
|
|
|
|
|
|
# --- (b) Flag on + session → incumbent kwarg present ---
|
|
|
|
class TestFlagOnSession:
|
|
def test_incumbent_kwarg_present_when_flag_on(self, tmp_path, monkeypatch):
|
|
"""(b) Flag on + session history → incumbent kwarg present
|
|
AND challenger_cache_rate is a concrete float."""
|
|
|
|
# Spy on rank_candidates to capture its keyword arguments.
|
|
class _Capturer:
|
|
def __init__(self):
|
|
self.kwargs = None
|
|
def __call__(self, rows, **kwargs):
|
|
self.kwargs = kwargs
|
|
from routing import rank_candidates as _real
|
|
return _real(rows, **kwargs)
|
|
|
|
capturer = _Capturer()
|
|
monkeypatch.setattr(dispatcher, "rank_candidates", capturer)
|
|
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
# challenger_cache_rate from config — a concrete float
|
|
monkeypatch.setattr(
|
|
dispatcher.cfg.objective, "incumbent_challenger_cache_rate", 0.90
|
|
)
|
|
# Set session context so route() resolves the incumbent.
|
|
# The /route endpoint doesn't set _current_session_key itself;
|
|
# the ContextVar is context-local and TestClient runs the sync
|
|
# handler on the calling thread, so a plain .set() here is
|
|
# visible to route().
|
|
dispatcher._current_session_key.set("ses_w6")
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
result = _route_via_client(client)
|
|
assert result["selected"] is not None
|
|
|
|
# Verify the incumbent and dial kwargs reached rank_candidates.
|
|
assert capturer.kwargs is not None, "rank_candidates was never called"
|
|
assert capturer.kwargs["incumbent"] is not None, "incumbent is None"
|
|
assert isinstance(capturer.kwargs["incumbent"], tuple) and len(capturer.kwargs["incumbent"]) == 2, \
|
|
f"incumbent is not a (provider, model_id) tuple: {capturer.kwargs['incumbent']}"
|
|
assert capturer.kwargs["challenger_cache_rate"] == pytest.approx(0.90), \
|
|
f"challenger_cache_rate != 0.90: {capturer.kwargs['challenger_cache_rate']}"
|
|
|
|
|
|
# --- (c) Exploration session gate ---
|
|
|
|
class TestExplorationGate:
|
|
def test_exploration_not_called_with_incumbent(self, tmp_path, monkeypatch):
|
|
"""(c.i) Exploration NOT called when incumbent is present.
|
|
|
|
With incumbency pricing on and session history, even if
|
|
`cfg.exploration.enabled` the gate rejects exploration.
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "max_tier", 3)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 0.5)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
# Set session context so route() resolves the incumbent.
|
|
dispatcher._current_session_key.set("ses_w6")
|
|
|
|
# Spy on exploration.choose — should NOT be called
|
|
choose_calls = []
|
|
original = dispatcher.exploration.choose
|
|
def _track(*a, **kw):
|
|
choose_calls.append(True)
|
|
return original(*a, **kw)
|
|
monkeypatch.setattr(dispatcher.exploration, "choose", _track)
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
result = _route_via_client(client)
|
|
assert result["selected"] is not None
|
|
assert len(choose_calls) == 0, (
|
|
"explore.choose should NOT be called when incumbent resolves"
|
|
)
|
|
|
|
def test_exploration_called_without_incumbent(self, tmp_path, monkeypatch):
|
|
"""(c.ii) Exploration IS called when there is no incumbent.
|
|
|
|
With the flag on but no session history, the `and incumbent is None`
|
|
gate allows exploration through (subject to its own epsilon/tier
|
|
checks).
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
# No session row → no incumbent
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "max_tier", 3)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 1.0)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
|
|
choose_calls = []
|
|
def _track(ranked, *args, **kwargs):
|
|
choose_calls.append(True)
|
|
# Return (ranked[0], True) so the test doesn't hang
|
|
return ranked[0] if ranked else None, True
|
|
monkeypatch.setattr(dispatcher.exploration, "choose", _track)
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
result = _route_via_client(client)
|
|
assert result["selected"] is not None
|
|
assert len(choose_calls) == 1, (
|
|
"explore.choose should be called when no incumbent"
|
|
)
|
|
|
|
|
|
# --- (d) Debug log emits incumbent identity ---
|
|
|
|
class TestDebugLog:
|
|
def test_debug_log_emits_incumbent_identity(self, tmp_path, monkeypatch, caplog):
|
|
"""(d) Debug log emits incumbent identity when present.
|
|
|
|
Verifies that the incumbent diagnostic block logs the session's
|
|
incumbent model and the challenger dial value.
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
_seed_measured_cache_rate(conn, "nw", "nw-model")
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
monkeypatch.setattr(
|
|
dispatcher.cfg.objective, "incumbent_challenger_cache_rate", 0.50
|
|
)
|
|
dispatcher._current_session_key.set("ses_w6")
|
|
|
|
caplog.set_level(logging.DEBUG, logger="llm_router")
|
|
# Ensure the llm_router logger accepts DEBUG so
|
|
# logs.enabled_for_debug() returns True.
|
|
import logging as _logging
|
|
_logging.getLogger("llm_router").setLevel(_logging.DEBUG)
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
_route_via_client(client)
|
|
|
|
# Check that the incumbent diagnostic appears in the logs
|
|
log_text = caplog.text
|
|
assert "incumbent" in log_text.lower(), (
|
|
"Debug log should contain incumbent diagnostic"
|
|
)
|
|
assert "nw/nw-model" in log_text, (
|
|
"Debug log should contain incumbent identity NW/nw-model"
|
|
)
|
|
# The per-row rank log carries the incumbent_pricing stamp (rate +
|
|
# source) on the incumbent's row — seeded rates make it 'measured'.
|
|
assert "incumbent_pricing" in log_text, (
|
|
"rank log should include the incumbent_pricing stamp"
|
|
)
|
|
assert "measured" in log_text, (
|
|
"stamp source should be 'measured' when rates were seeded"
|
|
)
|
|
|
|
def test_debug_log_identity_without_measured_rates(
|
|
self, tmp_path, monkeypatch, caplog
|
|
):
|
|
"""Fail-open: identity logs even when measured rates come back empty.
|
|
|
|
``_measured_cache_rates()`` returns ``{}`` on any failure or when the
|
|
pricing floor is not met. The incumbent identity line must still
|
|
fire (rates=None), and the rank log's stamp must fall back to
|
|
source='assumed'.
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
# Deliberately NO _seed_measured_cache_rate call here.
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", True)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
monkeypatch.setattr(
|
|
dispatcher.cfg.objective, "incumbent_challenger_cache_rate", 0.50
|
|
)
|
|
dispatcher._current_session_key.set("ses_w6")
|
|
|
|
caplog.set_level(logging.DEBUG, logger="llm_router")
|
|
import logging as _logging
|
|
_logging.getLogger("llm_router").setLevel(_logging.DEBUG)
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
_route_via_client(client)
|
|
|
|
log_text = caplog.text
|
|
assert "incumbent" in log_text.lower(), (
|
|
"identity must log even when measured rates are empty"
|
|
)
|
|
assert "nw/nw-model" in log_text, (
|
|
"identity must carry the incumbent model"
|
|
)
|
|
assert "incumbent_pricing" in log_text, (
|
|
"rank log should include the assumed-rate stamp"
|
|
)
|
|
assert "assumed" in log_text, (
|
|
"stamp source should be 'assumed' without measured rates"
|
|
)
|
|
|
|
def test_rank_log_has_no_incumbent_pricing_kwarg_when_flag_off(
|
|
self, tmp_path, monkeypatch, caplog
|
|
):
|
|
"""Flag off → rank log lines byte-identical (no incumbent_pricing kwarg).
|
|
|
|
The per-row kwarg is omitted when no stamp exists, so a flag-off run
|
|
emits the same rank log format as the pre-feature code.
|
|
"""
|
|
db_path = _make_db(tmp_path)
|
|
conn = _route_conn(db_path)
|
|
_seed_models_for_route(conn)
|
|
_seed_incumbent_session(conn)
|
|
_seed_measured_cache_rate(conn, "nw", "nw-model")
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.objective, "incumbent_cache_pricing", False)
|
|
monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test")
|
|
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
|
|
|
|
caplog.set_level(logging.DEBUG, logger="llm_router")
|
|
import logging as _logging
|
|
_logging.getLogger("llm_router").setLevel(_logging.DEBUG)
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
_route_via_client(client)
|
|
|
|
log_text = caplog.text
|
|
assert "rank" in log_text, "rank log should still fire"
|
|
assert "incumbent_pricing" not in log_text, (
|
|
"flag-off rank log must not gain an incumbent_pricing field"
|
|
)
|