349 lines
17 KiB
Python
349 lines
17 KiB
Python
"""Gates: blockieren hart, Stillstand → Pause, Coverage-Formel deterministisch."""
|
|
import asyncio
|
|
|
|
from backend import db, engine, graph
|
|
from .conftest import lauf_bis_ende, topic_anlegen
|
|
|
|
|
|
async def test_gate_blockiert_hart_und_stillstand_pausiert(monkeypatch):
|
|
"""Konsens liefert nie bestätigte Punkte → gate_korpus bleibt rot →
|
|
nach Runden-Cap/Stillstand pausiert der Lauf. Stufe 2 startet NIE."""
|
|
from backend import korpus
|
|
|
|
async def konsens_kaputt(task):
|
|
return engine.Ergebnis(daten={"bestaetigt": 0})
|
|
|
|
monkeypatch.setitem(engine.WORKER, "korpus.soll_konsens", konsens_kaputt)
|
|
topic = topic_anlegen()
|
|
run = await lauf_bis_ende(topic)
|
|
assert run["status"] == "paused"
|
|
assert "gate" in run["grund"] or "stillstand" in run["grund"]
|
|
# Kein einziger Task jenseits der Korpus-Stufe
|
|
g = graph.get()
|
|
spaeter = [k.id for k in g.knoten.values() if k.stufe != "korpus"]
|
|
qs = ",".join("?" for _ in spaeter)
|
|
assert db.one(f"SELECT id FROM tasks WHERE topic=? AND knoten IN ({qs}) "
|
|
"LIMIT 1", topic, *spaeter) is None
|
|
# Status-Marker wurde nie gesetzt
|
|
assert db.one("SELECT status FROM topics WHERE name=?", topic)["status"] == "neu"
|
|
|
|
|
|
async def test_stillstand_nur_ohne_neue_tasks():
|
|
"""Gleicher Fingerprint, aber NEUE Reparatur-Tasks → kein Pause-Abbruch;
|
|
erst wenn nichts Neues mehr entsteht, greift der Stillstand."""
|
|
import asyncio
|
|
from backend import engine, graph
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
g = graph.get()
|
|
kn = g.knoten["gate_guide"]
|
|
befund = [{"art": "redundanz", "item": "baustein:1", "detail": "x"}]
|
|
|
|
async def gate_lauf(runde, tasks):
|
|
tid = db.insert("tasks", run_id=run_id, topic=topic, knoten="gate_guide",
|
|
item=f"runde:{runde}", runde=runde, status="laufend")
|
|
t = dict(db.one("SELECT * FROM tasks WHERE id=?", tid))
|
|
engine._abschluss_schreiben(
|
|
g, t, kn, engine.Ergebnis(gate_status="rot", befunde=befund,
|
|
neue_tasks=tasks))
|
|
|
|
await gate_lauf(1, [{"knoten": "fix", "item": "r2:fix:1"}])
|
|
await gate_lauf(2, [{"knoten": "fix", "item": "r3:klaerung:1"}]) # NEU → ok
|
|
assert db.one("SELECT status FROM runs WHERE id=?", run_id)[
|
|
"status"] == "running"
|
|
await gate_lauf(3, [{"knoten": "fix", "item": "r3:klaerung:1"}]) # ignoriert
|
|
assert db.one("SELECT status FROM runs WHERE id=?", run_id)[
|
|
"status"] == "paused"
|
|
|
|
|
|
async def test_gate_rot_ohne_reparatur_pausiert(monkeypatch):
|
|
"""Gate rot ohne Reparatur-Rezepte darf nicht endlos drehen → Pause."""
|
|
from backend import korpus
|
|
|
|
async def gate_rot(task):
|
|
return engine.Ergebnis(gate_status="rot", befunde=[
|
|
{"art": "test", "item": "x", "detail": str(task["runde"])}])
|
|
|
|
monkeypatch.setitem(engine.WORKER, "korpus.gate", gate_rot)
|
|
topic = topic_anlegen()
|
|
run = await lauf_bis_ende(topic)
|
|
assert run["status"] == "paused"
|
|
|
|
|
|
async def test_coverage_formel():
|
|
topic = topic_anlegen()
|
|
run = await lauf_bis_ende(topic)
|
|
assert run["status"] == "done"
|
|
gesamt = db.one("SELECT COUNT(*) c FROM soll WHERE topic=? AND "
|
|
"status='bestaetigt'", topic)["c"]
|
|
gedeckt = db.one(
|
|
"SELECT COUNT(*) c FROM soll s WHERE s.topic=? AND s.status='bestaetigt' "
|
|
"AND (s.freispruch!='' OR EXISTS (SELECT 1 FROM atome a WHERE "
|
|
"a.soll_id=s.id AND a.status='aktiv' AND a.baustein_id IS NOT NULL))",
|
|
topic)["c"]
|
|
assert gesamt == gedeckt == 3 # Coverage 100 %, deterministisch gezählt
|
|
|
|
|
|
async def test_resume_mit_budget_nachschlag():
|
|
import asyncio
|
|
from backend import engine
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="budget", budget_tokens=100)
|
|
run_id2 = engine.lauf_starten(topic, budget=500)
|
|
assert run_id2 == run_id # gleicher Run wird fortgesetzt
|
|
assert db.one("SELECT budget_tokens FROM runs WHERE id=?", run_id)[
|
|
"budget_tokens"] == 500
|
|
engine.lauf_stoppen(topic)
|
|
with __import__("contextlib").suppress(asyncio.CancelledError):
|
|
await engine._laeufe[topic]
|
|
|
|
|
|
async def test_gate_guide_neuschreiben_bei_kaputten_markern():
|
|
"""Marker-Defekt + Fix erschöpft → Gate erzeugt Writer-Neuschreiben-Task;
|
|
veraltete Marker-Befunde zählen nicht mehr als offen."""
|
|
from backend import guide
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
kid = db.insert("kapitel", topic=topic, titel="K", ord=0)
|
|
zid = db.insert("lernziele", topic=topic, text="z")
|
|
bid = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B",
|
|
kapitel_id=kid, ord=0)
|
|
aid = db.insert("atome", topic=topic, titel="Atom X", status="aktiv",
|
|
baustein_id=bid, definition="d")
|
|
db.insert("sections", baustein_id=bid, text="Text ohne Marker.",
|
|
fix_versuche=3)
|
|
erg = await guide.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "runde:1", "payload": "{}"})
|
|
assert erg.gate_status == "rot"
|
|
writer_tasks = [t for t in erg.neue_tasks if t["knoten"] == "writer"]
|
|
assert writer_tasks and writer_tasks[0]["payload"]["neu"] is True
|
|
# Neuschreiben resettet den Fix-Cap und ersetzt den Text
|
|
await guide.writer({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"knoten": "writer",
|
|
"item": f"r2:baustein:{bid}",
|
|
"payload": db.j({"baustein_id": bid, "neu": True})})
|
|
sec = db.one("SELECT * FROM sections WHERE baustein_id=?", bid)
|
|
assert f"<!-- atom: {aid} -->" in sec["text"]
|
|
assert sec["fix_versuche"] == 0
|
|
|
|
|
|
async def test_eskalations_exit_beendet_karussell():
|
|
"""Nach Fix-Cap + Klärung + 1 Neuschreiben wird der Befund eskaliert —
|
|
das Gate blockt nicht ewig (Echtlauf: 5 Sections kreisten bis Runde 8)."""
|
|
from backend import guide
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
kid = db.insert("kapitel", topic=topic, titel="K", ord=0)
|
|
zid = db.insert("lernziele", topic=topic, text="z")
|
|
bid = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B",
|
|
kapitel_id=kid, ord=0)
|
|
aid = db.insert("atome", topic=topic, titel="Atom X", status="aktiv",
|
|
baustein_id=bid, definition="d")
|
|
db.insert("sections", baustein_id=bid,
|
|
text=f"<!-- atom: {aid} -->\nText.", fix_versuche=3)
|
|
db.insert("befunde", run_id=run_id, stufe="guide", art="falsch",
|
|
item=f"baustein:{bid}", detail="x", geklaert=1)
|
|
# 1 Neuschreiben wurde schon versucht:
|
|
db.insert("tasks", run_id=run_id, topic=topic, knoten="writer",
|
|
item=f"r9:baustein:{bid}", art="reparatur", status="fertig")
|
|
erg = await guide.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "runde:1", "payload": "{}"})
|
|
b = db.one("SELECT status FROM befunde WHERE item=?", f"baustein:{bid}")
|
|
assert b["status"] == "eskaliert"
|
|
assert not [t for t in erg.neue_tasks if t["knoten"] in ("fix", "writer")]
|
|
|
|
|
|
async def test_atom_echo_semantisch():
|
|
"""Fettzeile ohne Gedankenstrich umging den Regex-Check (Audit Z.294/297) —
|
|
der semantische Vergleich mit der Atom-Definition fängt sie."""
|
|
from backend import guide
|
|
topic = topic_anlegen()
|
|
zid = db.insert("lernziele", topic=topic, text="z")
|
|
kid = db.insert("kapitel", topic=topic, titel="K", ord=0)
|
|
bid = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B",
|
|
kapitel_id=kid, ord=0)
|
|
aid = db.insert("atome", topic=topic, titel="Pausenregeln", status="aktiv",
|
|
baustein_id=bid,
|
|
definition="Empfohlene Pausen sind Dehnen, Meditation und "
|
|
"kurze Spaziergänge an der frischen Luft.")
|
|
echo_text = (f"<!-- atom: {aid} -->\n"
|
|
"**Empfohlene Pausen sind Dehnen, Meditation und kurze "
|
|
"Spaziergänge an der frischen Luft.** Danach mehr. "
|
|
+ "Wort " * 40)
|
|
assert any(b["art"] == "atom_echo"
|
|
for b in guide._det_befunde(topic, bid, echo_text))
|
|
sauber = (f"<!-- atom: {aid} -->\n"
|
|
"Gute Pausen leben von Bewegung und bewusstem Abschalten. "
|
|
+ "Wort " * 40)
|
|
assert not any(b["art"] == "atom_echo"
|
|
for b in guide._det_befunde(topic, bid, sauber))
|
|
|
|
|
|
async def test_vorwaerts_ignoriert_verlinkte_erwaehnung():
|
|
from backend import guide
|
|
topic = topic_anlegen()
|
|
kid1 = db.insert("kapitel", topic=topic, titel="K1", ord=0)
|
|
kid2 = db.insert("kapitel", topic=topic, titel="K2", ord=1)
|
|
zid = db.insert("lernziele", topic=topic, text="z")
|
|
bid1 = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B1",
|
|
kapitel_id=kid1, ord=0)
|
|
bid2 = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B2",
|
|
kapitel_id=kid2, ord=0)
|
|
a1 = db.insert("atome", topic=topic, titel="Frühes Konzept", status="aktiv",
|
|
baustein_id=bid1)
|
|
db.insert("atome", topic=topic, titel="Langes Spätkonzept", status="aktiv",
|
|
baustein_id=bid2)
|
|
mit_link = (f"<!-- atom: {a1} -->\nSiehe [Langes Spätkonzept](#b2) später. "
|
|
+ "Wort " * 40)
|
|
assert not [b for b in guide._det_befunde(topic, bid1, mit_link)
|
|
if b["art"] == "vorwaerts"]
|
|
ohne_link = f"<!-- atom: {a1} -->\nDas Langes Spätkonzept kommt. " + "Wort " * 40
|
|
assert [b for b in guide._det_befunde(topic, bid1, ohne_link)
|
|
if b["art"] == "vorwaerts"]
|
|
|
|
|
|
async def test_gate_verwirft_soll_lose_atome(monkeypatch):
|
|
from backend import config, inventar
|
|
monkeypatch.setattr(config, "BEIFANG_MAX_QUOTE", 1.0) # Stop hier nicht Testziel
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
geprueft = db.insert("atome", topic=topic, titel="Beifang", status="aktiv",
|
|
soll_geprueft=1)
|
|
frisch = db.insert("atome", topic=topic, titel="Frisch", status="aktiv",
|
|
soll_geprueft=0)
|
|
await inventar.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"knoten": "gate_inventar", "item": "runde:1",
|
|
"payload": "{}"})
|
|
assert db.one("SELECT status FROM atome WHERE id=?", geprueft)[
|
|
"status"] == "verworfen"
|
|
assert db.one("SELECT status FROM atome WHERE id=?", frisch)[
|
|
"status"] == "aktiv" # ungeprüfte bleiben
|
|
|
|
|
|
async def test_befund_dedupe_erhaelt_klaerungsstand():
|
|
from backend.guide import befund_merken
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
assert befund_merken(run_id, "llm_pruefung", "luecke", "baustein:1",
|
|
"Detail X", 1) is True
|
|
db.execute("UPDATE befunde SET geklaert=1")
|
|
# gleiche Runde später: KEINE neue Zeile, geklaert bleibt
|
|
assert befund_merken(run_id, "llm_pruefung", "luecke", "baustein:1",
|
|
"Detail X", 2) is False
|
|
rows = db.query("SELECT geklaert FROM befunde WHERE art='luecke'")
|
|
assert len(rows) == 1 and rows[0]["geklaert"] == 1
|
|
# behoben → Regression darf neu aufmachen
|
|
db.execute("UPDATE befunde SET status='behoben'")
|
|
assert befund_merken(run_id, "llm_pruefung", "luecke", "baustein:1",
|
|
"Detail X", 3) is True
|
|
|
|
|
|
async def test_finale_det_befunde_blocken_nicht():
|
|
"""Eskalierter/freigesprochener Vorwärtsverweis hält das Gate nicht mehr rot."""
|
|
from backend import guide
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
kid1 = db.insert("kapitel", topic=topic, titel="K1", ord=0)
|
|
kid2 = db.insert("kapitel", topic=topic, titel="K2", ord=1)
|
|
zid = db.insert("lernziele", topic=topic, text="z")
|
|
bid1 = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B1",
|
|
kapitel_id=kid1, ord=0)
|
|
bid2 = db.insert("bausteine", topic=topic, ziel_id=zid, titel="B2",
|
|
kapitel_id=kid2, ord=0)
|
|
a1 = db.insert("atome", topic=topic, titel="Frühes Thema", status="aktiv",
|
|
baustein_id=bid1)
|
|
a2 = db.insert("atome", topic=topic, titel="Langes Spätkonzept",
|
|
status="aktiv", baustein_id=bid2)
|
|
db.insert("sections", baustein_id=bid1,
|
|
text=f"<!-- atom: {a1} -->\nDas Langes Spätkonzept kommt. "
|
|
+ "Wort " * 50)
|
|
db.insert("sections", baustein_id=bid2,
|
|
text=f"<!-- atom: {a2} -->\n" + "Wort " * 50)
|
|
db.insert("kapitel", topic=topic, titel="egal", ord=2)
|
|
erg1 = await guide.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "runde:1", "payload": "{}"})
|
|
vor = [b for b in erg1.befunde if b["art"] == "vorwaerts"]
|
|
assert vor # blockt zunächst
|
|
db.insert("befunde", run_id=run_id, stufe="guide", art="vorwaerts",
|
|
item=vor[0]["item"], detail=vor[0]["detail"][:500],
|
|
status="freispruch")
|
|
erg2 = await guide.gate({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"item": "runde:2", "payload": "{}"})
|
|
assert not [b for b in erg2.befunde if b["art"] == "vorwaerts"]
|
|
|
|
|
|
async def test_budget_stoppt_lauf():
|
|
topic = topic_anlegen()
|
|
run = await lauf_bis_ende(topic, budget=1)
|
|
# Fake-Calls kosten 0 Tokens — Budget greift nur mit echten Tokens.
|
|
# Simuliere Verbrauch und prüfe den Stopp-Pfad direkt:
|
|
if run["status"] == "done":
|
|
db.insert("events", run_id=run["id"], tok_in=5, tok_out=5)
|
|
from backend.ledger import BudgetErschoepft, budget_pruefen
|
|
import pytest
|
|
with pytest.raises(BudgetErschoepft):
|
|
budget_pruefen(run["id"])
|
|
|
|
|
|
async def test_stop_seed_fehlt(monkeypatch):
|
|
"""Vollständigkeit ist DAS Kernziel: ohne nützlich geladene Startquelle
|
|
PAUSE statt Mini-Guide (Lektion 103/106)."""
|
|
import pytest
|
|
from backend import config, korpus, llm
|
|
monkeypatch.setattr(config, "SEED_PFLICHT", True)
|
|
monkeypatch.setattr(config, "QUELLEN_MIN", 1)
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
db.insert("quellen", topic=topic, url="https://a.de", url_norm="a",
|
|
backend="ddgs", status="geladen", zweck="korpus")
|
|
for i in range(3):
|
|
db.insert("soll", topic=topic, punkt=f"P{i}", status="bestaetigt")
|
|
db.insert("tasks", run_id=run_id, topic=topic, knoten="gate_korpus",
|
|
item="runde:1", runde=1, status="fertig",
|
|
ergebnis=db.j({"bestaetigt": 3}))
|
|
with pytest.raises(llm.LaufPause, match="seed_fehlt"):
|
|
await korpus.gate({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"knoten": "gate_korpus", "item": "runde:2",
|
|
"payload": "{}"})
|
|
|
|
|
|
async def test_stop_zu_wenige_quellen(monkeypatch):
|
|
import pytest
|
|
from backend import config, korpus, llm
|
|
monkeypatch.setattr(config, "SEED_PFLICHT", False)
|
|
monkeypatch.setattr(config, "QUELLEN_MIN", 5)
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
db.insert("quellen", topic=topic, url="https://a.de", url_norm="a",
|
|
backend="ddgs", status="geladen", zweck="korpus")
|
|
for i in range(3):
|
|
db.insert("soll", topic=topic, punkt=f"P{i}", status="bestaetigt")
|
|
db.insert("tasks", run_id=run_id, topic=topic, knoten="gate_korpus",
|
|
item="runde:1", runde=1, status="fertig",
|
|
ergebnis=db.j({"bestaetigt": 3}))
|
|
with pytest.raises(llm.LaufPause, match="quellen"):
|
|
await korpus.gate({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"knoten": "gate_korpus", "item": "runde:2",
|
|
"payload": "{}"})
|
|
|
|
|
|
async def test_stop_beifang_quote():
|
|
""">50 % der Atome als Beifang verworfen → Soll zu eng → PAUSE."""
|
|
import pytest
|
|
from backend import inventar, llm
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
s = db.insert("soll", topic=topic, punkt="P", status="bestaetigt",
|
|
belege=db.j([]))
|
|
behalten = db.insert("atome", topic=topic, titel="Gut", status="aktiv",
|
|
soll_id=s, soll_geprueft=1)
|
|
for i in range(2):
|
|
db.insert("atome", topic=topic, titel=f"Beifang{i}", status="aktiv",
|
|
soll_geprueft=1)
|
|
with pytest.raises(llm.LaufPause, match="beifang"):
|
|
await inventar.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"knoten": "gate_inventar", "item": "runde:1",
|
|
"payload": "{}"})
|
|
assert db.one("SELECT status FROM atome WHERE id=?", behalten)[
|
|
"status"] == "aktiv"
|