187 lines
9.6 KiB
Python
187 lines
9.6 KiB
Python
"""Auftrags-Lebenszyklus (Guide): eingefrorener Wortlaut, Urteil je Runde,
|
||
Fehlalarm-Tod (einstimmig), Runden-Cap → eskaliert, Fund nur bei leerer Liste."""
|
||
|
||
import db
|
||
import fake_agents
|
||
import guide
|
||
import llm
|
||
import pytest
|
||
from config import AUFTRAG_RUNDEN_MAX
|
||
from conftest import run_anlegen, topic_anlegen
|
||
|
||
|
||
def _baustein(topic, lang_extra=""):
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
||
b_id = db.insert("bausteine", topic=topic, ziel_id=ziel, titel="B", ord=0, status="neu")
|
||
a = db.insert("atome", topic=topic, titel="A", typ="begriff", definition="d",
|
||
status="neu", baustein_id=b_id)
|
||
lang = f"<!-- atom: {a} | A -->\n" + ("Wort " * 60).strip() + lang_extra
|
||
db.insert("sections", baustein_id=b_id, stage="pruefer", text_lang=lang,
|
||
text_kompakt="- p")
|
||
return db.one("SELECT * FROM bausteine WHERE id=?", (b_id,))
|
||
|
||
|
||
def _ctx(topic):
|
||
ctx = llm.Kontext(run_anlegen(topic), topic, "minimax")
|
||
ctx.ebene = "guide"
|
||
return ctx
|
||
|
||
|
||
def test_migration_befunde_json_zu_auftraegen():
|
||
import sqlite3
|
||
con = sqlite3.connect(":memory:")
|
||
con.execute("CREATE TABLE sections(baustein_id INTEGER PRIMARY KEY,"
|
||
" stage TEXT NOT NULL DEFAULT 'writer', text_kompakt TEXT DEFAULT '',"
|
||
" text_lang TEXT DEFAULT '', befunde TEXT DEFAULT '[]',"
|
||
" qa_hash TEXT DEFAULT '', fix_versuche INTEGER NOT NULL DEFAULT 0)")
|
||
con.execute("INSERT INTO sections(baustein_id, befunde) VALUES(7, ?)", (db.j([
|
||
"KRITISCH (falsch): Zahl stimmt nicht",
|
||
"KRITISCH (luecke): Atom 5 fehlt",
|
||
"KRITISCH (fachlich_falsch): Claim X",
|
||
"KRITISCH (marker_fehlend): Marker 5 fehlt", # messen-Art → verwerfen
|
||
"KRITISCH: Marker für Atom 5 fehlt — exakt einfügen.", # det → verwerfen
|
||
"Langtext, Absatz 2 hat 130 Wörter (Regel: 40–90): teilen."]),))
|
||
db._init_schema(con)
|
||
zeilen = con.execute("SELECT art, detail, quelle FROM auftraege").fetchall()
|
||
assert sorted(z[0] for z in zeilen) == ["falsch", "falsch", "luecke"]
|
||
assert ("falsch", "Claim X", "qa") in zeilen # fachlich_falsch → falsch, quelle qa
|
||
# Spalte weg + idempotent (zweiter Lauf legt nichts doppelt an)
|
||
assert not any(s[1] == "befunde" for s in con.execute("PRAGMA table_info(sections)"))
|
||
db._init_schema(con)
|
||
assert con.execute("SELECT COUNT(*) FROM auftraege").fetchone()[0] == 3
|
||
|
||
|
||
async def test_urteil_statusuebergaenge_und_keine_neuen_funde(monkeypatch):
|
||
topic = topic_anlegen("urteil")
|
||
b = _baustein(topic)
|
||
a1 = db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="D1")
|
||
a2 = db.insert("auftraege", baustein_id=b["id"], art="stil", detail="D2")
|
||
a3 = db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="D3")
|
||
urteile = {1: "behoben", 2: "kein_mangel", 3: "faktenbasis"}
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer-Urteil",
|
||
lambda p: [{"auftrag": i, "urteil": u} for i, u in urteile.items()])
|
||
# Fund-Modus darf bei offenen Aufträgen NIE laufen
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer",
|
||
lambda p: pytest.fail("Fund-Modus trotz offener Aufträge"))
|
||
await guide._auftraege_urteilen(_ctx(topic), b,
|
||
db.one("SELECT * FROM sections WHERE baustein_id=?",
|
||
(b["id"],)),
|
||
guide._offene_auftraege(b["id"]))
|
||
stat = {r["id"]: r["status"] for r in db.query("SELECT * FROM auftraege")}
|
||
assert stat == {a1: "behoben", a2: "kein_mangel", a3: "eskaliert"}
|
||
|
||
|
||
async def test_kein_mangel_nur_einstimmig(monkeypatch):
|
||
topic = topic_anlegen("split")
|
||
b = _baustein(topic)
|
||
au = db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="D")
|
||
votes = iter(["kein_mangel", "offen", "offen"]) # Stimme 1 ≠ Stimme 2 (+Ersatz)
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer-Urteil",
|
||
lambda p: [{"auftrag": 1, "urteil": next(votes)}])
|
||
await guide._auftraege_urteilen(_ctx(topic), b,
|
||
db.one("SELECT * FROM sections WHERE baustein_id=?",
|
||
(b["id"],)),
|
||
guide._offene_auftraege(b["id"]))
|
||
r = db.one("SELECT * FROM auftraege WHERE id=?", (au,))
|
||
assert r["status"] == "offen" and r["runden"] == 1 # uneins → offen, Runde zählt
|
||
|
||
|
||
async def test_runden_cap_eskaliert_und_wird_nicht_geroutet(monkeypatch):
|
||
topic = topic_anlegen("cap")
|
||
b = _baustein(topic)
|
||
au = db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="zäh",
|
||
runden=AUFTRAG_RUNDEN_MAX - 1)
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer-Urteil",
|
||
lambda p: [{"auftrag": 1, "urteil": "offen"}])
|
||
ctx = _ctx(topic)
|
||
await guide._auftraege_urteilen(ctx, b,
|
||
db.one("SELECT * FROM sections WHERE baustein_id=?",
|
||
(b["id"],)),
|
||
guide._offene_auftraege(b["id"]))
|
||
assert db.one("SELECT status FROM auftraege WHERE id=?", (au,))["status"] == "eskaliert"
|
||
# fakten_konflikt wird in reparieren nicht geroutet → kein Fortschritt = False
|
||
db.update("sections", "baustein_id", b["id"], stage="done")
|
||
bewegt = await guide.reparieren(ctx, [{"art": "fakten_konflikt",
|
||
"item": str(b["id"]), "detail": "zäh"}])
|
||
assert bewegt is False
|
||
assert db.one("SELECT stage FROM sections WHERE baustein_id=?",
|
||
(b["id"],))["stage"] == "done"
|
||
|
||
|
||
async def test_fund_dedup_gegen_kein_mangel(monkeypatch):
|
||
"""Ein totgestimmter Fehlalarm darf nicht als frische Zeile auferstehen."""
|
||
topic = topic_anlegen("wiedergaenger")
|
||
b = _baustein(topic)
|
||
db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="Fehlalarm X",
|
||
status="kein_mangel")
|
||
gesehen = {}
|
||
def finder(prompt):
|
||
gesehen["negativ"] = "Fehlalarm X" in prompt
|
||
return {"befunde": [{"art": "falsch", "detail": "Fehlalarm X"},
|
||
{"art": "luecke", "detail": "Neu Y"}]}
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer", finder)
|
||
await guide._auftraege_finden(_ctx(topic), b,
|
||
db.one("SELECT * FROM sections WHERE baustein_id=?",
|
||
(b["id"],)))
|
||
assert gesehen["negativ"] # Negativ-Liste stand im Prompt
|
||
offen = guide._offene_auftraege(b["id"])
|
||
assert [a["detail"] for a in offen] == ["Neu Y"]
|
||
|
||
|
||
async def test_fix_stil_optimistisch_falsch_via_urteil(monkeypatch):
|
||
topic = topic_anlegen("fixarten")
|
||
b = _baustein(topic)
|
||
stil = db.insert("auftraege", baustein_id=b["id"], art="stil", detail="glätten")
|
||
falsch = db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="Zahl prüfen")
|
||
sec = db.one("SELECT * FROM sections WHERE baustein_id=?", (b["id"],))
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Fix",
|
||
lambda p: {"kompakt": "- p", "lang": sec["text_lang"] + " neu"})
|
||
# Re-Check-Urteil lässt den falsch-Auftrag OFFEN → nicht optimistisch geschlossen
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Guide-Pruefer-Urteil",
|
||
lambda p: [{"auftrag": 1, "urteil": "offen"}])
|
||
db.update("sections", "baustein_id", b["id"], stage="fix")
|
||
assert await guide._stage_fix(_ctx(topic), b) == "done"
|
||
assert db.one("SELECT status FROM auftraege WHERE id=?", (stil,))["status"] == "behoben"
|
||
assert db.one("SELECT status FROM auftraege WHERE id=?", (falsch,))["status"] == "offen"
|
||
|
||
|
||
def test_qa_auftrag_dedup_und_eskalation():
|
||
topic = topic_anlegen("qapfad")
|
||
b = _baustein(topic)
|
||
guide._qa_auftrag(b["id"], "Claim A")
|
||
guide._qa_auftrag(b["id"], "Claim A") # exakter Dedup
|
||
zeilen = db.query("SELECT * FROM auftraege WHERE baustein_id=?", (b["id"],))
|
||
assert len(zeilen) == 1 and zeilen[0]["quelle"] == "qa"
|
||
# Re-Bestätigung auf kein_mangel-Zeile: Richter uneins → eskaliert
|
||
db.update("auftraege", "id", zeilen[0]["id"], status="kein_mangel")
|
||
guide._qa_auftrag(b["id"], "Claim A")
|
||
assert db.one("SELECT status FROM auftraege WHERE id=?",
|
||
(zeilen[0]["id"],))["status"] == "eskaliert"
|
||
|
||
|
||
async def test_urteilsrunde_ohne_text_ist_bewegt(monkeypatch):
|
||
"""Statusübergänge zählen als Fortschritt — sonst bricht auto_loop reine
|
||
Urteilsrunden als Stillstand ab, obwohl Aufträge geschlossen wurden."""
|
||
topic = topic_anlegen("bewegt2")
|
||
b = _baustein(topic)
|
||
db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="D")
|
||
db.update("sections", "baustein_id", b["id"], stage="done")
|
||
# Fake-Urteil: behoben (Default-Handler); Text bleibt unverändert
|
||
bewegt = await guide.reparieren(_ctx(topic), [
|
||
{"art": "auftrag_offen", "item": str(b["id"]), "detail": "(falsch) D"}])
|
||
assert bewegt is True
|
||
assert db.query("SELECT * FROM auftraege WHERE status='offen'") == []
|
||
|
||
|
||
async def test_freeze_urteilt_weiter(monkeypatch):
|
||
"""Fix-Cap friert nur Stil-Fixes ein — Aufträge werden trotzdem geurteilt."""
|
||
topic = topic_anlegen("freeze")
|
||
b = _baustein(topic)
|
||
db.update("sections", "baustein_id", b["id"],
|
||
stage="done", fix_versuche=guide.FIX_MAX_VERSUCHE)
|
||
db.insert("auftraege", baustein_id=b["id"], art="falsch", detail="D")
|
||
bewegt = await guide.reparieren(_ctx(topic), [
|
||
{"art": "auftrag_offen", "item": str(b["id"]), "detail": "(falsch) D"}])
|
||
assert bewegt is True # geurteilt (behoben) trotz Freeze
|
||
assert db.one("SELECT COUNT(*) n FROM auftraege WHERE status='behoben'")["n"] == 1
|