150 lines
5.5 KiB
Python
150 lines
5.5 KiB
Python
"""The classifier circuit breaker actually breaking the circuit.
|
|
|
|
``_last_classifier_failure`` was written by ``_record_failure`` and read by
|
|
nothing: the cascade gated only its *cloud* step, and on a per-provider refusal
|
|
state (``_provider_refusal_since``). So a dead local classifier was re-dialled
|
|
on every request. A refused connection is cheap; a hung or VPN-black-holed
|
|
Ollama costs
|
|
``classifier.timeout_seconds`` — 120s on the reference deployment — per request,
|
|
for as long as the outage lasts.
|
|
|
|
Every test here asserts on whether the local client was *constructed*, because
|
|
that is the thing whose absence saves the 120s. Asserting on the returned
|
|
Classification alone would pass even if the router had waited out the timeout
|
|
first.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
from openai import APIConnectionError
|
|
|
|
import dispatcher
|
|
import session_cache
|
|
from dispatcher import Classification
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _closed_circuit(monkeypatch):
|
|
"""Start every test with an empty cache and a closed circuit."""
|
|
session_cache.clear()
|
|
monkeypatch.setattr(dispatcher, "_last_classifier_failure", 0.0)
|
|
dispatcher._provider_refusal_since.clear()
|
|
# No session key -> cascade steps 1 and 2 are impossible, so a skipped
|
|
# local attempt lands on the static fallback and nothing touches the DB.
|
|
token = dispatcher._current_session_key.set(None)
|
|
yield
|
|
dispatcher._current_session_key.reset(token)
|
|
dispatcher._provider_refusal_since.clear()
|
|
session_cache.clear()
|
|
|
|
|
|
def _stub_local(monkeypatch, *, answer=None, boom=None):
|
|
"""Stub the local classifier; return the list that records constructions."""
|
|
built: list[int] = []
|
|
|
|
def _client():
|
|
built.append(1)
|
|
return object()
|
|
|
|
def _once(client, system_prompt, user_content):
|
|
if boom is not None:
|
|
raise boom
|
|
return answer
|
|
|
|
monkeypatch.setattr(dispatcher, "_classifier_client", _client)
|
|
monkeypatch.setattr(dispatcher, "_classify_once", _once)
|
|
return built
|
|
|
|
|
|
def _boom() -> Exception:
|
|
"""A connection failure of the shape a stopped Ollama produces."""
|
|
return APIConnectionError(request=None) # type: ignore[arg-type]
|
|
|
|
|
|
def _answer() -> Classification:
|
|
return Classification(
|
|
task_category="coding_general",
|
|
task_tier=2,
|
|
required_context_tokens=100,
|
|
confidence=0.9,
|
|
source="classifier",
|
|
)
|
|
|
|
|
|
def test_local_classifier_is_skipped_during_the_cooldown_window(monkeypatch):
|
|
"""The whole point: an open circuit must not dial the dead endpoint."""
|
|
built = _stub_local(monkeypatch, answer=_answer())
|
|
monkeypatch.setattr(dispatcher, "_last_classifier_failure", time.time())
|
|
|
|
got = dispatcher.classify("do a thing", None)
|
|
|
|
assert built == [], "local classifier client built inside the cooldown window"
|
|
assert got.source == "fallback"
|
|
|
|
|
|
def test_a_failure_opens_the_window_for_the_following_request(monkeypatch):
|
|
"""One failure, then the next request skips the local attempt entirely."""
|
|
built = _stub_local(monkeypatch, boom=_boom())
|
|
first = dispatcher.classify("do a thing", None)
|
|
assert first.source == "fallback"
|
|
assert built == [1], "the first attempt should have been made"
|
|
assert dispatcher._last_classifier_failure > 0.0
|
|
|
|
dispatcher.classify("do another thing", None)
|
|
assert built == [1], "second request re-dialled a classifier known to be down"
|
|
|
|
|
|
def test_a_skipped_attempt_does_not_extend_the_window(monkeypatch):
|
|
"""A skip must not re-stamp the failure clock, or the circuit never closes.
|
|
|
|
``_classify_cascade`` records a failure on entry. If a *skipped* local
|
|
attempt walked that same path, every request during an outage would push
|
|
the deadline forward and the local classifier would never be re-probed
|
|
while traffic kept flowing — a permanent outage rather than a backoff.
|
|
"""
|
|
_stub_local(monkeypatch, answer=_answer())
|
|
cooldown = dispatcher.cfg.classifier.cooldown_seconds
|
|
opened_at = time.time() - (cooldown - 1)
|
|
monkeypatch.setattr(dispatcher, "_last_classifier_failure", opened_at)
|
|
|
|
for _ in range(3):
|
|
dispatcher.classify("do a thing", None)
|
|
|
|
# abs=, not the default relative tolerance: 1e-6 of a ~1.8e9 epoch is
|
|
# ±1800s, which would swallow the very re-stamp this asserts against.
|
|
assert dispatcher._last_classifier_failure == pytest.approx(opened_at, abs=0.5), (
|
|
"a skipped attempt re-stamped the failure clock, so the window "
|
|
"never expires while traffic flows"
|
|
)
|
|
|
|
|
|
def test_the_window_expires_and_the_local_classifier_is_retried(monkeypatch):
|
|
"""A backoff that never re-probes is an outage, not a backoff."""
|
|
built = _stub_local(monkeypatch, answer=_answer())
|
|
cooldown = dispatcher.cfg.classifier.cooldown_seconds
|
|
monkeypatch.setattr(
|
|
dispatcher, "_last_classifier_failure", time.time() - (cooldown + 1)
|
|
)
|
|
|
|
got = dispatcher.classify("do a thing", None)
|
|
|
|
assert built == [1], "the cooldown expired but the local attempt never happened"
|
|
assert got.source == "classifier"
|
|
|
|
|
|
def test_a_success_closes_the_circuit(monkeypatch):
|
|
"""After an answer, the next request must not inherit a stale window."""
|
|
built = _stub_local(monkeypatch, answer=_answer())
|
|
cooldown = dispatcher.cfg.classifier.cooldown_seconds
|
|
monkeypatch.setattr(
|
|
dispatcher, "_last_classifier_failure", time.time() - (cooldown + 1)
|
|
)
|
|
|
|
dispatcher.classify("do a thing", None)
|
|
assert dispatcher._last_classifier_failure == 0.0
|
|
|
|
dispatcher.classify("do another thing", None)
|
|
assert built == [1, 1], "a healthy classifier was skipped"
|