bfb0ff7c00
The router previously recorded only completions (energy_observations), not the routing decisions behind them, so 'how routing is performing' was not answerable from data. This adds: - route_decisions table + idempotent ensure_route_decisions (guarded CREATE TABLE IF NOT EXISTS, never regenerates a live DB) gated by logging.log_route_decisions; every decision kind (route/dispatch/chat/ passthrough/local-vision) is persisted best-effort via persist_route_decision (never fails a request; only session_key, never session_dir). The table is ensured on the write path (mirroring proficiency_store._write -> ensure_columns) so a live DB that predates the feature migrates safely. - metrics.py aggregator moved quota_burn/scoring_coverage in from the dispatcher (breaking a would-be circular import) and adds recent_decisions/per_model/verdict_mix/top_proficiency; /health now imports them and GET /metrics exposes the 7-key JSON (window-bounded, loopback-only, no auth). - observed_at indexes on energy_observations/verifications.
661 lines
22 KiB
Python
661 lines
22 KiB
Python
"""Tests for the route_decisions table, its inline-create helper, and the gate.
|
|
|
|
The monitoring TUI (see .omo/plans/router-monitoring-tui.md) needs a record of
|
|
every routing decision — which model was picked and why — that survives in the
|
|
datbase rather than only in the journal. This file pins the three pieces todo #1
|
|
adds:
|
|
|
|
- the `route_decisions` table in schema.sql (columns and the guarded index),
|
|
- `dispatcher.ensure_route_decisions(conn)` — the idempotent inline-create
|
|
helper that is the *only* way the table appears on a live router.db (the
|
|
live DB is never recreated; schema.sql alone is CREATE TABLE IF NOT EXISTS
|
|
and silently does nothing to an existing DB),
|
|
- the `logging.log_route_decisions` config gate.
|
|
|
|
The database tests follow the same offline temp-DB pattern as
|
|
tests/test_chat_completions.py: a throwaway SQLite file seeded from schema.sql,
|
|
never the live router.db.
|
|
"""
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
import dispatcher
|
|
from dispatcher import Classification, app
|
|
import config
|
|
import metrics
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCHEMA_SQL = (ROOT / "schema.sql").read_text()
|
|
|
|
ROUTE_DECISIONS_COLUMNS = [
|
|
"id",
|
|
"observed_at",
|
|
"kind",
|
|
"task_category",
|
|
"task_tier",
|
|
"required_context_tokens",
|
|
"confidence",
|
|
"classifier_ms",
|
|
"classification_source",
|
|
"latency_tolerance",
|
|
"candidates_considered",
|
|
"selected_model",
|
|
"selected_provider",
|
|
"runner_up_models",
|
|
"est_cost_usd",
|
|
"est_proficiency",
|
|
"rejected_reason",
|
|
"session_key",
|
|
"tools",
|
|
"images",
|
|
"json_mode",
|
|
"streamed",
|
|
]
|
|
|
|
|
|
def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
|
|
row = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
|
|
(table,),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
|
|
def _index_exists(conn: sqlite3.Connection, index: str) -> bool:
|
|
row = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='index' AND name=?",
|
|
(index,),
|
|
).fetchone()
|
|
return row is not None
|
|
|
|
|
|
def _schema_minus_route_decisions() -> str:
|
|
"""schema.sql with the route_decisions block removed, for the failure case."""
|
|
lines = []
|
|
skipping = False
|
|
for line in SCHEMA_SQL.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("CREATE TABLE IF NOT EXISTS route_decisions"):
|
|
skipping = True
|
|
continue
|
|
if skipping:
|
|
# The create ends at the closing paren + semicolon of the table
|
|
# statement. Anything still in the table body is skipped.
|
|
if stripped == ");":
|
|
skipping = False
|
|
continue
|
|
if "idx_route_decisions_observed" in line:
|
|
continue
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
|
|
|
|
# --- schema round-trips -----------------------------------------------------
|
|
|
|
|
|
def test_schema_defines_route_decisions_table():
|
|
"""schema.sql declares the table, so a fresh DB from it has it already."""
|
|
assert "CREATE TABLE IF NOT EXISTS route_decisions" in SCHEMA_SQL
|
|
|
|
|
|
def test_schema_index_is_guarded():
|
|
"""The observed_at index must be IF NOT EXISTS so re-applying is a no-op."""
|
|
assert "idx_route_decisions_observed" in SCHEMA_SQL
|
|
assert (
|
|
"CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
|
|
"ON route_decisions (observed_at)" in SCHEMA_SQL
|
|
)
|
|
|
|
|
|
# --- happy path: fresh DB from full schema ----------------------------------
|
|
|
|
|
|
def test_fresh_schema_already_has_table(tmp_path):
|
|
conn = sqlite3.connect(tmp_path / "fresh.db")
|
|
conn.executescript(SCHEMA_SQL)
|
|
assert _table_exists(conn, "route_decisions")
|
|
for col in ROUTE_DECISIONS_COLUMNS:
|
|
assert col in {r[1] for r in conn.execute("PRAGMA table_info(route_decisions)")}
|
|
assert _index_exists(conn, "idx_route_decisions_observed")
|
|
conn.close()
|
|
|
|
|
|
def test_ensure_route_decisions_is_idempotent(tmp_path):
|
|
"""Fresh DB already has the table; calling the helper twice no-ops."""
|
|
conn = sqlite3.connect(tmp_path / "idem.db")
|
|
conn.executescript(SCHEMA_SQL)
|
|
# Seed some rows in another table so we can prove nothing is dropped.
|
|
conn.execute(
|
|
"INSERT INTO models (model_id, provider, last_updated) "
|
|
"VALUES ('m1', 'neuralwatt', '2026-01-01T00:00:00+00:00')"
|
|
)
|
|
conn.commit()
|
|
|
|
dispatcher.ensure_route_decisions(conn) # first call
|
|
dispatcher.ensure_route_decisions(conn) # second call: must no-op cleanly
|
|
|
|
assert _table_exists(conn, "route_decisions")
|
|
count = conn.execute("SELECT COUNT(*) FROM models").fetchone()[0]
|
|
assert count == 1 # pre-existing rows survived
|
|
conn.close()
|
|
|
|
|
|
# --- failure path: pre-existing DB WITHOUT the table ------------------------
|
|
|
|
|
|
def test_ensure_route_decisions_adds_table_without_dropping_rows(tmp_path):
|
|
"""A DB that predates the table gets it added; existing rows survive."""
|
|
conn = sqlite3.connect(tmp_path / "old.db")
|
|
conn.executescript(_schema_minus_route_decisions())
|
|
assert not _table_exists(conn, "route_decisions")
|
|
# A row in a genuinely existing table, to prove it survives the upgrade.
|
|
conn.execute(
|
|
"INSERT INTO models (model_id, provider, last_updated) "
|
|
"VALUES ('legacy', 'neuralwatt', '2026-01-01T00:00:00+00:00')"
|
|
)
|
|
conn.commit()
|
|
|
|
dispatcher.ensure_route_decisions(conn)
|
|
|
|
assert _table_exists(conn, "route_decisions")
|
|
assert _index_exists(conn, "idx_route_decisions_observed")
|
|
legacy = conn.execute(
|
|
"SELECT model_id FROM models WHERE model_id='legacy'"
|
|
).fetchone()
|
|
assert legacy is not None # the existing row was not dropped
|
|
conn.close()
|
|
|
|
|
|
def test_ensure_route_decisions_allows_insert(tmp_path):
|
|
"""After the helper runs, the table actually accepts the documented shape."""
|
|
conn = sqlite3.connect(tmp_path / "insert.db")
|
|
conn.executescript(_schema_minus_route_decisions())
|
|
dispatcher.ensure_route_decisions(conn)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO route_decisions (
|
|
observed_at, kind, task_category, task_tier,
|
|
required_context_tokens, confidence, classifier_ms,
|
|
classification_source, latency_tolerance, candidates_considered,
|
|
selected_model, selected_provider, runner_up_models,
|
|
est_cost_usd, est_proficiency, rejected_reason, session_key,
|
|
tools, images, json_mode, streamed
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
"2026-01-01T00:00:00+00:00", "route", "coding_general", 2, 500,
|
|
0.95, 1868, "classifier", "interactive", 8, "deepseek-v4-flash",
|
|
"neuralwatt", '[{"model_id": "gemma-4-31b", "provider": "neuralwatt"}]',
|
|
0.00016296, 1.0, None, "sess-hash", 0, 0, 0, 1,
|
|
),
|
|
)
|
|
conn.commit()
|
|
kind = conn.execute(
|
|
"SELECT kind FROM route_decisions WHERE selected_model='deepseek-v4-flash'"
|
|
).fetchone()
|
|
assert kind is not None and kind[0] == "route"
|
|
conn.close()
|
|
|
|
|
|
def test_persist_ensure_on_write_fixes_live_db_missing_table(tmp_path, monkeypatch):
|
|
"""A live router.db without route_decisions gets it on the WRITE path.
|
|
|
|
F4 scope-fidelity regression: ``persist_route_decision`` INSERTs without
|
|
ever calling ``ensure_route_decisions``, so if a live DB lacks the table
|
|
and the module-load startup hook did not run (a test harness, a process
|
|
that calls persist first, a future lazy-import refactor), every decision
|
|
row is silently swallowed (the INSERT raises no-such-table) and /metrics'
|
|
recent_decisions 500s. Mirroring proficiency_store._write -> ensure_columns,
|
|
the migration must be guaranteed on the write path too, not only at module
|
|
load. The temp DB is deliberately the schema-minus-route_decisions shape —
|
|
a DB that predates the feature.
|
|
"""
|
|
db_path = tmp_path / "live-no-table.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(_schema_minus_route_decisions())
|
|
assert not _table_exists(conn, "route_decisions")
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
|
|
monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
|
|
|
|
# The write itself. Best-effort means it must not raise even on a missing
|
|
# table; then the row must actually land and the table must exist.
|
|
dispatcher.persist_route_decision(
|
|
"route",
|
|
classification=Classification(
|
|
task_category="coding_general", task_tier=2,
|
|
required_context_tokens=100, confidence=0.9,
|
|
),
|
|
latency_tolerance="interactive",
|
|
)
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
assert _table_exists(conn, "route_decisions"), \
|
|
"the migration must run on the write path so the table exists"
|
|
rows = conn.execute(
|
|
"SELECT kind FROM route_decisions WHERE kind='route'"
|
|
).fetchall()
|
|
assert len(rows) == 1, "the decision row must persist once the table exists"
|
|
|
|
# /metrics / recent_decisions against the same live-DB shape must not 500:
|
|
# reading must succeed now that the table exists.
|
|
rec = metrics.recent_decisions(conn)
|
|
assert len(rec) == 1
|
|
assert rec[0]["kind"] == "route"
|
|
conn.close()
|
|
|
|
|
|
# --- config gate ------------------------------------------------------------
|
|
|
|
|
|
def test_log_route_decisions_gate_defaults_on():
|
|
"""The key is declared and defaults to on, matching the config.yaml value."""
|
|
cfg = config.load_config(str(ROOT / "config.yaml"))
|
|
assert cfg.logging.log_route_decisions is True
|
|
|
|
|
|
def test_config_has_log_route_decisions_key():
|
|
"""The strict config accepts the key — it must be declared or load fails."""
|
|
assert "log_route_decisions" in (ROOT / "config.yaml").read_text()
|
|
|
|
|
|
# =============================================================================
|
|
# Todo #2: persist_route_decision wired into every decision path.
|
|
# =============================================================================
|
|
|
|
CHEAP = "cheap-model"
|
|
DEAR = "dear-model"
|
|
|
|
|
|
class FakeResponse:
|
|
"""Just enough of requests.Response for the dispatcher's provider calls."""
|
|
|
|
def __init__(self, payload=None, *, status_code=200, lines=None):
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
self._lines = lines or []
|
|
self.text = json.dumps(self._payload)
|
|
self.closed = False
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
def iter_lines(self, decode_unicode=False):
|
|
yield from self._lines
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def _completion(model, content="hello there"):
|
|
return {
|
|
"id": "chatcmpl-dec-1",
|
|
"model": model,
|
|
"choices": [
|
|
{"message": {"role": "assistant", "content": content},
|
|
"finish_reason": "stop"}
|
|
],
|
|
"usage": {"prompt_tokens": 31, "completion_tokens": 12},
|
|
"energy": {"energy_kwh": 5.0e-05, "carbon_g_co2eq": 2.4e-03},
|
|
"cost": {"request_cost_usd": 4.0e-04},
|
|
}
|
|
|
|
|
|
_STREAM_LINES = [
|
|
'data: {"id":"chatcmpl-stream-dec","choices":[{"delta":{"content":"hel"}}]}',
|
|
"",
|
|
'data: {"id":"chatcmpl-stream-dec","choices":[{"delta":{"content":"lo"},'
|
|
'"finish_reason":"stop"}],"usage":{"prompt_tokens":31,'
|
|
'"completion_tokens":9}}',
|
|
"",
|
|
"data: [DONE]",
|
|
"",
|
|
]
|
|
|
|
|
|
def _messages(text="write me a function"):
|
|
return [{"role": "user", "content": text}]
|
|
|
|
|
|
def _image_messages():
|
|
return [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "text", "text": "what is in this image?"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
|
|
],
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def decision_router(tmp_path, monkeypatch):
|
|
"""A routable dispatcher over a throwaway DB; nothing dials out."""
|
|
db_path = tmp_path / "decisions.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(SCHEMA_SQL)
|
|
for model_id, completion_price, vision in (
|
|
(CHEAP, 0.30, 1),
|
|
(DEAR, 9.00, 0),
|
|
):
|
|
conn.execute(
|
|
"""
|
|
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 (?, 'neuralwatt', ?, 2, 262128, 192500, 16384, ?, ?,
|
|
?, 1, 'standard', 'default', 'full', 'public', 'active',
|
|
'2026-08-22T00:00:00+00:00')
|
|
""",
|
|
(model_id, model_id, completion_price / 3, completion_price, vision),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
|
|
monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
|
|
monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
|
|
# ensure the gate is on for the happy-path tests (default, but pin it).
|
|
monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
|
|
|
|
calls = []
|
|
|
|
def fake_post(url, headers=None, json=None, stream=False, timeout=None):
|
|
calls.append({"url": url, "body": json, "stream": stream})
|
|
if stream:
|
|
return FakeResponse(lines=_STREAM_LINES)
|
|
return FakeResponse(_completion(json["model"]))
|
|
|
|
monkeypatch.setattr(dispatcher.requests, "post", fake_post)
|
|
monkeypatch.setattr(
|
|
dispatcher, "classify",
|
|
lambda task, context: Classification(
|
|
task_category="coding_general", task_tier=2,
|
|
required_context_tokens=100, confidence=0.9,
|
|
),
|
|
)
|
|
|
|
class _Raw:
|
|
text = json.dumps(_completion(CHEAP))
|
|
|
|
class _Completions:
|
|
@property
|
|
def with_raw_response(self):
|
|
return self
|
|
|
|
def create(self, **kwargs):
|
|
return _Raw()
|
|
|
|
class _FakeClient:
|
|
chat = type("_Chat", (), {"completions": _Completions()})()
|
|
|
|
monkeypatch.setattr(dispatcher, "_provider_client", lambda provider: _FakeClient())
|
|
yield TestClient(app), db_path
|
|
|
|
|
|
def _rows(db_path):
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute(
|
|
"SELECT * FROM route_decisions ORDER BY id ASC"
|
|
).fetchall()
|
|
conn.close()
|
|
return rows
|
|
|
|
|
|
def _drop_cheap(db_path):
|
|
"""Make CHEAP ineligible so DEAR (or nothing) remains."""
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("UPDATE models SET tier = 1 WHERE model_id = ?", (CHEAP,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
# --- happy paths ------------------------------------------------------------
|
|
|
|
|
|
def test_route_endpoint_persists_one_row(decision_router):
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/route", json={"task": "write me a function"}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["kind"] == "route"
|
|
assert r["selected_model"] == CHEAP
|
|
assert r["selected_provider"] == "neuralwatt"
|
|
assert r["classification_source"] == "classifier"
|
|
assert r["task_category"] == "coding_general"
|
|
assert r["task_tier"] == 2
|
|
assert r["latency_tolerance"] == "interactive"
|
|
assert r["session_key"] is None
|
|
assert "session_dir" not in r.keys()
|
|
|
|
|
|
def test_route_endpoint_override_has_classifier_ms_null(decision_router):
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/route",
|
|
json={
|
|
"task": "x",
|
|
"task_category": "coding_refactor",
|
|
"task_tier": 3,
|
|
"required_context_tokens": 5000,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["classification_source"] == "override"
|
|
assert r["classifier_ms"] is None, "an override never consulted the classifier"
|
|
assert r["task_category"] == "coding_refactor"
|
|
|
|
|
|
def test_dispatch_endpoint_persists_one_row(decision_router):
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/dispatch",
|
|
json={"task": "write me a function",
|
|
"task_category": "coding_general",
|
|
"task_tier": 2,
|
|
"required_context_tokens": 100},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["kind"] == "dispatch"
|
|
assert r["selected_model"] == CHEAP
|
|
assert r["classification_source"] == "override"
|
|
|
|
|
|
def test_routed_chat_persists_one_row(decision_router):
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "auto", "messages": _messages()},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["kind"] == "chat"
|
|
assert r["selected_model"] == CHEAP
|
|
assert r["selected_provider"] == "neuralwatt"
|
|
assert r["classification_source"] == "classifier"
|
|
assert r["session_key"] is not None
|
|
# The session key is a hash — never a directory, never content.
|
|
assert len(r["session_key"]) == 16
|
|
assert "/" not in (r["session_key"] or "")
|
|
|
|
|
|
def test_routed_chat_reroute_keeps_classifier_source(decision_router):
|
|
"""Re-routing for measured context must not persist source='override'."""
|
|
client, db_path = decision_router
|
|
# >100 measured tokens (measured = chars/3), so chat_completions reroutes.
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "auto", "messages": _messages("refactor " + "x " * 600)},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
assert rows[0]["classification_source"] == "classifier", \
|
|
"the re-route's source='override' must not overwrite the classifier's"
|
|
assert rows[0]["classifier_ms"] is not None
|
|
|
|
|
|
def test_streamed_routed_chat_persists_one_row(decision_router):
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "auto", "messages": _messages(), "stream": True},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
assert rows[0]["kind"] == "chat"
|
|
assert rows[0]["selected_model"] == CHEAP
|
|
assert rows[0]["streamed"] == 1
|
|
|
|
|
|
def test_passthrough_persists_one_row_with_no_nameerror(decision_router):
|
|
"""The pre-existing pass-through NameError must stay gone, and a row lands."""
|
|
client, db_path = decision_router
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": DEAR, "messages": _messages()},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["kind"] == "passthrough"
|
|
assert r["selected_model"] == DEAR
|
|
assert r["selected_provider"] == "neuralwatt"
|
|
assert r["classification_source"] is None
|
|
assert r["session_key"] is not None
|
|
|
|
|
|
def test_local_vision_success_persists_one_local_row(decision_router, monkeypatch):
|
|
client, db_path = decision_router
|
|
_drop_cheap(db_path)
|
|
monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
|
|
|
|
def fake_post(url, headers=None, json=None, stream=False, timeout=None):
|
|
return FakeResponse(
|
|
{"choices": [{"message": {"role": "assistant",
|
|
"content": "local caption"},
|
|
"finish_reason": "stop"}]}
|
|
)
|
|
|
|
monkeypatch.setattr(dispatcher.requests, "post", fake_post)
|
|
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "auto", "messages": _image_messages()},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1, "one decision, and only one: the local_vision row"
|
|
r = rows[0]
|
|
assert r["kind"] == "local_vision"
|
|
assert r["selected_model"] == dispatcher.cfg.local_vision.model
|
|
assert r["selected_provider"] == "local"
|
|
assert r["rejected_reason"] is None
|
|
assert r["images"] == 1
|
|
|
|
|
|
def test_no_candidate_422_still_persists_a_rejection_row(decision_router):
|
|
client, db_path = decision_router
|
|
# No vision cloud candidate, local fallback disabled -> 422.
|
|
_drop_cheap(db_path)
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "auto", "messages": _image_messages()},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
rows = _rows(db_path)
|
|
assert len(rows) == 1
|
|
r = rows[0]
|
|
assert r["kind"] == "chat"
|
|
assert r["selected_model"] is None
|
|
assert r["rejected_reason"] is not None
|
|
assert "vision" in r["rejected_reason"]
|
|
|
|
|
|
# --- failure modes: best-effort, config-gated -------------------------------
|
|
|
|
|
|
def test_gate_off_writes_nothing_but_routing_still_200(decision_router):
|
|
client, db_path = decision_router
|
|
dispatcher.cfg.logging.log_route_decisions = False
|
|
try:
|
|
resp = client.post("/route", json={"task": "write me a function"})
|
|
assert resp.status_code == 200
|
|
finally:
|
|
dispatcher.cfg.logging.log_route_decisions = True
|
|
|
|
assert _rows(db_path) == [], "the gate off must leave the table untouched"
|
|
|
|
|
|
def _raise_on_route_decisions_insert(db_path):
|
|
real_db = dispatcher._db
|
|
|
|
class GuardingConn:
|
|
def __init__(self, conn):
|
|
self._conn = conn
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._conn, name)
|
|
|
|
def execute(self, sql, parameters=()):
|
|
if isinstance(sql, str) and "INSERT INTO route_decisions" in sql:
|
|
raise sqlite3.OperationalError("database is locked")
|
|
return self._conn.execute(sql, parameters)
|
|
|
|
def wrapped_db():
|
|
return GuardingConn(real_db())
|
|
|
|
return wrapped_db
|
|
|
|
|
|
def test_db_write_failure_never_fails_routing(decision_router, monkeypatch):
|
|
"""A locked/read-only DB must not error the request; persistence is best-effort."""
|
|
client, db_path = decision_router
|
|
monkeypatch.setattr(dispatcher, "_db", _raise_on_route_decisions_insert(db_path))
|
|
|
|
resp = client.post("/route", json={"task": "write me a function"})
|
|
assert resp.status_code == 200, "a failed decision write must never fail routing"
|
|
|
|
# And the same holds for a routed completion.
|
|
resp2 = client.post(
|
|
"/v1/chat/completions", json={"model": "auto", "messages": _messages()}
|
|
)
|
|
assert resp2.status_code == 200
|