serversdown
e631797187
Observed live: an overheated MI50 returns a single char repeated ("?????") as a
successful 200, which neither the timeout nor the exception fallback catches — so
a degraded GPU would silently save capped garbage gists. Validate each summary
call's output: flag text (>=24 non-space chars) whose most-common non-whitespace
char exceeds 50%, raise DegenerateOutput, and let the existing retry->cloud
fallback handle it. Real prose (top char <20%) won't false-positive; short output
is exempt; cloud garbage raises rather than looping.
Tests: _looks_degenerate flags repeated-char / passes real prose / ignores short;
degenerate MI50 output falls back to cloud; cloud garbage raises. 177 pass.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015yrEb5qpPGv2FjyxrB7LLk
143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
"""Summary consolidation: MI50 length cap, fast-fail, and cloud fallback.
|
|
|
|
Everything is stubbed — no real backend is touched. These drive the behavior of
|
|
`summary._summarize_text`: try the primary backend a bounded number of times with
|
|
a capped generation length, and fall back to cloud if the primary keeps failing.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from lyra import summary
|
|
|
|
|
|
@pytest.fixture
|
|
def calls(monkeypatch):
|
|
"""Capture every llm.complete call; per-test behavior via `fake.responder`."""
|
|
recorded: list[dict] = []
|
|
|
|
def fake_complete(messages, backend="local", model=None,
|
|
max_tokens=None, timeout=None):
|
|
recorded.append({"backend": backend, "max_tokens": max_tokens, "timeout": timeout})
|
|
return fake_complete.responder(backend)
|
|
|
|
fake_complete.responder = lambda backend: "gist"
|
|
monkeypatch.setattr(summary.llm, "complete", fake_complete)
|
|
monkeypatch.setattr(summary.time, "sleep", lambda *_: None) # instant backoff
|
|
return types.SimpleNamespace(recorded=recorded, fake=fake_complete)
|
|
|
|
|
|
def _set_key(monkeypatch, key="sk-test"):
|
|
monkeypatch.setattr(summary.config, "load",
|
|
lambda: types.SimpleNamespace(openai_api_key=key))
|
|
|
|
|
|
def test_falls_back_to_cloud_after_mi50_attempts(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
|
|
def responder(backend):
|
|
if backend == "mi50":
|
|
raise RuntimeError("Request timed out.")
|
|
return "cloud-gist"
|
|
calls.fake.responder = responder
|
|
|
|
out = summary._summarize_text("transcript", "mi50")
|
|
|
|
assert out == "cloud-gist"
|
|
assert [c["backend"] for c in calls.recorded] == ["mi50", "mi50", "cloud"]
|
|
|
|
|
|
def test_no_fallback_when_backend_is_cloud(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
calls.fake.responder = lambda backend: (_ for _ in ()).throw(RuntimeError("boom"))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
summary._summarize_text("t", "cloud")
|
|
|
|
# Cloud is already the primary: retry it, but never a redundant fallback.
|
|
assert [c["backend"] for c in calls.recorded] == ["cloud", "cloud"]
|
|
|
|
|
|
def test_no_fallback_without_openai_key(calls, monkeypatch):
|
|
_set_key(monkeypatch, key="")
|
|
calls.fake.responder = lambda backend: (_ for _ in ()).throw(RuntimeError("mi50 down"))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
summary._summarize_text("t", "mi50")
|
|
|
|
assert [c["backend"] for c in calls.recorded] == ["mi50", "mi50"]
|
|
|
|
|
|
def test_caps_length_and_timeout_on_every_call(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
|
|
def responder(backend):
|
|
if backend == "mi50":
|
|
raise RuntimeError("nope")
|
|
return "cloud-gist"
|
|
calls.fake.responder = responder
|
|
|
|
summary._summarize_text("t", "mi50")
|
|
|
|
assert calls.recorded
|
|
for c in calls.recorded:
|
|
assert c["max_tokens"] == summary.SUMMARY_MAX_TOKENS
|
|
assert c["timeout"] == summary.SUMMARY_TIMEOUT
|
|
|
|
|
|
def test_happy_path_uses_primary_only(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
calls.fake.responder = lambda backend: "mi50-gist"
|
|
|
|
out = summary._summarize_text("t", "mi50")
|
|
|
|
assert out == "mi50-gist"
|
|
assert [c["backend"] for c in calls.recorded] == ["mi50"] # no retries, no fallback
|
|
|
|
|
|
# --- degenerate ("?" garbage) output guard: a wedged local model returns junk as
|
|
# a successful 200, so treat it as a failure and fall back to cloud. ---
|
|
|
|
def test_looks_degenerate_flags_repeated_char():
|
|
assert summary._looks_degenerate("?" * 60) is True
|
|
assert summary._looks_degenerate("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!") is True
|
|
|
|
|
|
def test_looks_degenerate_passes_real_prose():
|
|
gist = ("Brian sat down at the Meadows 1/3 in seat 6 with two straddles active; "
|
|
"he tagged a seat-3 calling station and finished the session up 240.")
|
|
assert summary._looks_degenerate(gist) is False
|
|
|
|
|
|
def test_looks_degenerate_ignores_short_output():
|
|
# Too short to judge — don't false-positive a terse-but-valid reply.
|
|
assert summary._looks_degenerate("ok") is False
|
|
|
|
|
|
def test_degenerate_mi50_output_falls_back_to_cloud(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
|
|
def responder(backend):
|
|
if backend == "mi50":
|
|
return "?" * 200 # garbage-as-200, not an exception
|
|
return "a real cloud gist of the session, diverse and coherent."
|
|
calls.fake.responder = responder
|
|
|
|
out = summary._summarize_text("transcript", "mi50")
|
|
|
|
assert "cloud gist" in out
|
|
assert [c["backend"] for c in calls.recorded] == ["mi50", "mi50", "cloud"]
|
|
|
|
|
|
def test_degenerate_cloud_output_raises_no_infinite_loop(calls, monkeypatch):
|
|
_set_key(monkeypatch)
|
|
calls.fake.responder = lambda backend: "?" * 200 # every backend returns garbage
|
|
|
|
with pytest.raises(Exception):
|
|
summary._summarize_text("t", "mi50")
|
|
|
|
# mi50 x2, then one cloud fallback that's also garbage -> give up, no loop.
|
|
assert [c["backend"] for c in calls.recorded] == ["mi50", "mi50", "cloud"]
|