123 lines
5.8 KiB
Python
123 lines
5.8 KiB
Python
"""Verdichtung: Fusion nur im Schnitt beider Panel-Stimmen, Anker-Vererbung,
|
|
verdichtet-Flag, Coverage bleibt erhalten."""
|
|
from backend import config, db, inventar
|
|
from .conftest import topic_anlegen
|
|
|
|
|
|
def _welt(topic, n_vorteile=3):
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
sid = db.insert("soll", topic=topic, punkt="Vorteile erklären",
|
|
status="bestaetigt")
|
|
ids = []
|
|
for i in range(n_vorteile):
|
|
aid = db.insert("atome", topic=topic, titel=f"Vorteil Nr{i}",
|
|
definition=f"Definition des Vorteils Nummer {i} hier.",
|
|
status="aktiv", soll_id=sid)
|
|
db.insert("anker", atom_id=aid, quelle_id=i % 2 + 1, start=i * 100,
|
|
ende=i * 100 + 50, zitat=f"z{i}")
|
|
ids.append(aid)
|
|
return run_id, sid, ids
|
|
|
|
|
|
def _task(run_id, topic, sid):
|
|
return {"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": f"r1:verd:soll:{sid}", "payload": db.j({"soll_id": sid})}
|
|
|
|
|
|
async def test_fusion_mit_anker_vererbung():
|
|
topic = topic_anlegen()
|
|
run_id, sid, ids = _welt(topic)
|
|
erg = await inventar.verdichtung(_task(run_id, topic, sid))
|
|
assert erg.daten["nachher"] == 1 # Fake gruppiert alle "Vorteil"-Atome
|
|
kopf = db.one("SELECT id FROM atome WHERE soll_id=? AND status='aktiv'", sid)
|
|
assert db.one("SELECT COUNT(*) c FROM anker WHERE atom_id=?",
|
|
kopf["id"])["c"] == 3 # alle Anker geerbt
|
|
verworfen = db.query("SELECT merged_into FROM atome WHERE soll_id=? AND "
|
|
"status='verworfen'", sid)
|
|
assert len(verworfen) == 2
|
|
assert db.one("SELECT verdichtet FROM soll WHERE id=?", sid)["verdichtet"] == 1
|
|
# Coverage bleibt: Punkt hat weiter ≥1 aktives Atom
|
|
assert db.one("SELECT id FROM atome WHERE soll_id=? AND status='aktiv'", sid)
|
|
|
|
|
|
async def test_nur_einstimmige_paare_fusionieren(monkeypatch):
|
|
"""Stimme A gruppiert {1,2,3}, Stimme B nur {1,2} → fusioniert wird nur
|
|
das einstimmige Paar (1,2)."""
|
|
topic = topic_anlegen()
|
|
run_id, sid, ids = _welt(topic)
|
|
antworten = iter([
|
|
f"===EINHEIT===\nIDS: {ids[0]},{ids[1]},{ids[2]}",
|
|
f"===EINHEIT===\nIDS: {ids[0]},{ids[1]}\n===EINHEIT===\nIDS: {ids[2]}",
|
|
])
|
|
|
|
async def zwei_stimmen(**kw):
|
|
return next(antworten)
|
|
from backend import llm
|
|
monkeypatch.setattr(llm, "call", zwei_stimmen)
|
|
erg = await inventar.verdichtung(_task(run_id, topic, sid))
|
|
assert erg.daten["fusionen"] == 1
|
|
assert erg.daten["nachher"] == 2 # {0,1} fusioniert, 2 bleibt allein
|
|
|
|
|
|
async def test_flag_verhindert_zweiten_lauf():
|
|
topic = topic_anlegen()
|
|
run_id, sid, ids = _welt(topic)
|
|
await inventar.verdichtung(_task(run_id, topic, sid))
|
|
erg2 = await inventar.verdichtung({**_task(run_id, topic, sid),
|
|
"item": "r2:verd"})
|
|
assert erg2.daten.get("skip") is True
|
|
|
|
|
|
async def test_facetten_check_erzeugt_fenster_task(tmp_path):
|
|
"""Beleg-Stelle ohne Atom in der Nähe → luecke-Task mit gezieltem Fenster;
|
|
nach LUECKEN_RUNDEN_MAX Versuchen → eskaliert (dokumentierte Grenze)."""
|
|
from backend import config as cfg
|
|
topic = topic_anlegen()
|
|
run_id = db.insert("runs", topic=topic, status="running")
|
|
weit = "Anfangssatz mit Atomnähe. " + "Füllwort " * 2000 + \
|
|
"Die Namensherkunft stammt von der Tomatenuhr aus Italien."
|
|
pfad = tmp_path / "q.md"
|
|
pfad.write_text(weit, encoding="utf-8")
|
|
qid = db.insert("quellen", topic=topic, url="https://x.de/a",
|
|
url_norm="https://x.de/a", status="geladen",
|
|
snapshot=str(pfad))
|
|
sid = db.insert("soll", topic=topic, punkt="Herkunft erklären",
|
|
status="bestaetigt", verdichtet=1,
|
|
belege=db.j([{"quelle": qid,
|
|
"zitat": "Die Namensherkunft stammt von der "
|
|
"Tomatenuhr aus Italien."}]))
|
|
aid = db.insert("atome", topic=topic, titel="Anfang", status="aktiv",
|
|
soll_id=sid)
|
|
db.insert("anker", atom_id=aid, quelle_id=qid, start=0, ende=20, zitat="z")
|
|
erg = await inventar.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "runde:1", "payload": "{}"})
|
|
fac = [t for t in erg.neue_tasks if t["knoten"] == "luecke"
|
|
and t["payload"].get("fenster")]
|
|
assert fac, "Facetten-Lücke muss gezielten Fenster-Task erzeugen"
|
|
assert fac[0]["payload"]["fenster"]["quelle_id"] == qid
|
|
assert any(b["art"] == "facette" for b in erg.befunde)
|
|
# Cap: 2 erfolglose Versuche simulieren → Eskalation statt Endlosschleife
|
|
key = fac[0]["item"].split(":", 1)[1]
|
|
for r in (2, 3):
|
|
db.insert("tasks", run_id=run_id, topic=topic, knoten="luecke",
|
|
item=f"r{r}:{key}", status="fertig")
|
|
erg2 = await inventar.gate({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"item": "runde:2", "payload": "{}"})
|
|
assert not any(b["art"] == "facette" for b in erg2.befunde)
|
|
assert db.one("SELECT status FROM befunde WHERE art='facette'")[
|
|
"status"] == "eskaliert"
|
|
|
|
|
|
async def test_gate_erzeugt_verdichtungs_task():
|
|
topic = topic_anlegen()
|
|
run_id, sid, ids = _welt(topic)
|
|
erg = await inventar.gate({"run_id": run_id, "topic": topic, "runde": 1,
|
|
"item": "runde:1", "payload": "{}"})
|
|
verd = [t for t in erg.neue_tasks if t["knoten"] == "verdichtung"]
|
|
assert verd and verd[0]["payload"]["soll_id"] == sid
|
|
# nach Verdichtung: kein neuer Task mehr
|
|
db.update("soll", "id=?", (sid,), verdichtet=1)
|
|
erg2 = await inventar.gate({"run_id": run_id, "topic": topic, "runde": 2,
|
|
"item": "runde:2", "payload": "{}"})
|
|
assert not [t for t in erg2.neue_tasks if t["knoten"] == "verdichtung"]
|