187 lines
7.8 KiB
Python
187 lines
7.8 KiB
Python
"""Budget-Stopp (hart) und Infra-Fail-closed (Lauf-Pause statt Teilergebnis)."""
|
|
|
|
import asyncio
|
|
|
|
import agents
|
|
import ledger
|
|
import llm
|
|
import pytest
|
|
from conftest import run_anlegen, topic_anlegen
|
|
|
|
|
|
async def test_budget_stoppt_hart(monkeypatch):
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic, budget=120) # Fake-Call kostet 150 Tokens
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
with pytest.raises(ledger.BudgetErschoepft):
|
|
await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
assert ledger.verbraucht(run) >= 120 # der Verbrauch bleibt sichtbar
|
|
|
|
|
|
async def test_infra_pause_statt_fail_open(monkeypatch):
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
|
|
async def immer_429(key, prompt, timeout, **kw):
|
|
return agents.AgentErgebnis(1, "", "HTTP 429: rate limited")
|
|
|
|
monkeypatch.setattr(agents, "run_agent", immer_429)
|
|
monkeypatch.setattr(llm, "INFRA_BACKOFF_BASE", 0.01)
|
|
with pytest.raises(llm.LaufPause):
|
|
await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
stati = [e["status"] for e in
|
|
__import__("db").query("SELECT status FROM events WHERE run_id=?", (run,))]
|
|
assert stati and all(s == "infra" for s in stati) # jeder Versuch im Ledger
|
|
|
|
|
|
async def test_budget_vorab_stoppt_ohne_neuen_call(monkeypatch):
|
|
"""Erschöpftes Budget → nächster call() wirft, BEVOR ein Agent-Call startet."""
|
|
import db
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic, budget=100)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
# Budget künstlich überschreiten (ohne Agent-Call), Event-Zahl merken
|
|
ledger.log_call(run, ebene="test", stage="seed", status="ok",
|
|
tokens={"input": 200, "output": 0})
|
|
vorher = db.one("SELECT COUNT(*) AS n FROM events WHERE run_id=?", (run,))["n"]
|
|
|
|
async def darf_nicht_laufen(key, prompt, timeout, **kw):
|
|
raise AssertionError("Agent-Call trotz erschöpftem Budget gestartet")
|
|
|
|
monkeypatch.setattr(agents, "run_agent", darf_nicht_laufen)
|
|
with pytest.raises(ledger.BudgetErschoepft):
|
|
await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
nachher = db.one("SELECT COUNT(*) AS n FROM events WHERE run_id=?", (run,))["n"]
|
|
assert nachher == vorher # keine neue Ledger-Zeile = kein Call
|
|
|
|
|
|
async def test_hedge_gewinner_trotz_haupt_timeout(monkeypatch):
|
|
"""Haupt-Call läuft in Timeout, Zwilling (-h) liefert → Ergebnis statt Pause."""
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
monkeypatch.setattr(llm, "HEDGE_NACH_S", 0.05)
|
|
|
|
async def haupt_stallt(key, prompt, timeout, **kw):
|
|
if key.endswith("-h"): # Zwilling liefert gültiges JSON
|
|
return agents.AgentErgebnis(0, "[]", "")
|
|
await asyncio.sleep(10) # Haupt hängt → wird nach Timeout gecancelt
|
|
return agents.AgentErgebnis(0, "[]", "")
|
|
|
|
monkeypatch.setattr(agents, "run_agent", haupt_stallt)
|
|
# timeout klein halten, damit der Haupt-Task zügig als Timeout endet
|
|
monkeypatch.setattr(llm, "timeout_fuer", lambda *a, **k: 0.2)
|
|
res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
assert res == [] # Zwilling hat geliefert, kein LaufPause
|
|
|
|
|
|
async def test_hedge_verlierer_zaehlt_im_budget(monkeypatch):
|
|
"""Beide Hedge-Calls liefern → zwei Ledger-Zeilen, Verlierer-Tokens im Budget."""
|
|
import db
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
monkeypatch.setattr(llm, "HEDGE_NACH_S", 0.05)
|
|
monkeypatch.setattr(llm, "timeout_fuer", lambda *a, **k: 0.4) # schwelle=0.2
|
|
|
|
async def haupt_leer_zwilling_ok(key, prompt, timeout, **kw):
|
|
if key.endswith("-h"): # Zwilling startet bei 0.2, liefert bei ~0.35
|
|
await asyncio.sleep(0.15)
|
|
return agents.AgentErgebnis(0, "[]", "")
|
|
await asyncio.sleep(0.25) # Haupt: fertig, aber leer (nicht ok) → Verlierer
|
|
return agents.AgentErgebnis(1, "", "leer", {"input": 40, "output": 20})
|
|
|
|
monkeypatch.setattr(agents, "run_agent", haupt_leer_zwilling_ok)
|
|
res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
assert res == []
|
|
zeilen = db.query("SELECT status FROM events WHERE run_id=?", (run,))
|
|
assert any(z["status"] == "hedge" for z in zeilen) # Verlierer geloggt
|
|
assert ledger.verbraucht(run) >= 60 # Verlierer-Tokens (40+20) sichtbar
|
|
|
|
|
|
async def test_spawn_kill_bei_cancel(monkeypatch):
|
|
"""Task-Cancel (Hedge-Verlierer/Pause) killt den CLI-Prozess, kein Zombie."""
|
|
import os
|
|
monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False)
|
|
monkeypatch.setattr(agents, "_ram_gate", lambda key: _true())
|
|
task = asyncio.ensure_future(
|
|
agents._spawn("t-cancel", ["sleep", "30"], None, 30))
|
|
for _ in range(200): # warten bis der Prozess registriert ist
|
|
if agents._prozesse:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
proc = next(iter(agents._prozesse.values()))
|
|
pid = proc.pid
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
for _ in range(200): # Kill ist async
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
with pytest.raises(ProcessLookupError):
|
|
os.kill(pid, 0) # Prozess ist tot
|
|
assert not agents._prozesse # aus der Registry entfernt
|
|
|
|
|
|
async def _true():
|
|
return True
|
|
|
|
|
|
async def test_opencode_timeout_loggt_tokens(monkeypatch):
|
|
"""opencode-Timeout: Tokens der Session werden geloggt, nicht als 0 verworfen."""
|
|
monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False)
|
|
|
|
async def spawn_timeout(*a, **kw):
|
|
raise asyncio.TimeoutError
|
|
|
|
monkeypatch.setattr(agents, "_spawn", spawn_timeout)
|
|
monkeypatch.setattr(agents, "_opencode_tokens",
|
|
lambda key: {"input": 5000, "output": 3000})
|
|
res = await agents._opencode("t-oc", "prompt", 10, "minimax/M", "none")
|
|
assert res.err == "timeout" and not res.ok
|
|
assert res.tokens == {"input": 5000, "output": 3000} # nicht 0
|
|
|
|
|
|
def test_gather_nur_in_llm():
|
|
"""Rückfall-Guard: bare asyncio.gather darf NUR in llm.py stehen (sonst
|
|
laufen Geschwister-Tasks bei LaufPause/Budget weiter — nutze llm.alle)."""
|
|
from pathlib import Path
|
|
backend = Path(__file__).resolve().parent.parent / "backend"
|
|
treffer = []
|
|
for p in backend.glob("*.py"):
|
|
if p.name == "llm.py":
|
|
continue
|
|
for i, zeile in enumerate(p.read_text(encoding="utf-8").splitlines(), 1):
|
|
if "asyncio.gather" in zeile:
|
|
treffer.append(f"{p.name}:{i}")
|
|
assert not treffer, f"bare asyncio.gather außerhalb llm.py: {treffer}"
|
|
|
|
|
|
async def test_inhaltsfehler_kein_laufabbruch(monkeypatch):
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "test"
|
|
|
|
async def unsinn(key, prompt, timeout, **kw):
|
|
return agents.AgentErgebnis(0, "kein json", "")
|
|
|
|
monkeypatch.setattr(agents, "run_agent", unsinn)
|
|
res = await llm.call(ctx, stage="soll", template="Korpus-Soll",
|
|
werte={"topic": topic, "quelle": "q", "text": "x"}, erwartet=list)
|
|
assert res is None # begrenzte Restarts, dann None — kein Absturz, keine Pause
|