update

tests/test_auftraege.py

@@ -0,0 +1,186 @@
+"""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, braucht=db.j([]))
+    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