6krrt/tests/test_incumbent_routing.py

"""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 datetime import datetime, timezone
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.

    Rows are stamped "now", not a fixed date: _measured_cache_rates only reads
    rows inside objective.cache_rate_window_hours (168), so a literal date
    ages out of the window and the incumbent silently falls back to the
    assumed rate. That is how this fixture started failing on 2026-09-20.
    """
    observed_at = datetime.now(timezone.utc).isoformat()
    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', ?)""",
            (model_id, provider, prompt_tokens, cached_tokens, observed_at),
        )
    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"
        )