update

backend/bausteine.py

@@ -15,8 +15,10 @@ import logging
 import math
 import shutil
 import subprocess
+import time
 from pathlib import Path
 
+import database as db
 from agents import kill_process, cancel_scope, clear_scope
 from config import KONSENS_GRACE, RECHERCHE_GRACE, KONSENS_MAX_RUNDEN, DEFAULT_PROVIDER
 from fsutil import atomic_write_text, atomic_write_json
@@ -39,6 +41,13 @@ SUBBAUSTEIN_MAX = 40
 # Einstufen ist billig (kurzes Urteil, keine Websuche) → größere Pakete, weniger Dateien/Agenten.
 STUFE_CHUNK = 100
 
+# Recherche-Loop: 5 Agenten je Runde, Runden bis volle Crawl-Abdeckung / 0 neue / Zeit-Kappe.
+RECHERCHE_AGENTEN = 5
+RECHERCHE_QUORUM = 3        # je Runde mind. so viele gültige Slot-Ergebnisse
+RECHERCHE_KAPPE = 1800      # Loop-Gesamt-Sekunden (Notbremse, 30 min)
+SUBBAUSTEIN_KAPPE = 900     # Subbaustein-Finde-Loop je Chunk (15 min)
+KONSOLIDIERUNG_CHUNK = 150  # Konsolidierung ab so vielen Kandidaten chunken
+
 log = logging.getLogger("creator.bausteine")
 
 _bausteine_progress: dict[str, str] = {}
@@ -83,8 +92,12 @@ def _crawl_fertig(topic: str) -> bool:
 _STUFEN = ("einfach", "mittel", "schwer")
 
 
-def subbausteine_titel(topic: str, baustein: str) -> list[str]:
-    """Subbaustein-Titel eines Bausteins aus der Sidecar (leer, wenn keine)."""
+async def subbausteine_titel(topic: str, baustein: str) -> list[str]:
+    """Subbaustein-Titel eines Bausteins — DB-first (Konsens), Fallback Sidecar-Datei."""
+    rows = [s["sub_titel"] for s in await db.list_subbausteine(topic, _norm_titel(baustein))
+            if s["status"] == "konsens" and s["sub_titel"]]
+    if rows:
+        return rows
     sc = _json_datei(subbausteine_path(topic))
     if not isinstance(sc, dict):
         return []
@@ -94,8 +107,11 @@ def subbausteine_titel(topic: str, baustein: str) -> list[str]:
     ]
 
 
-def lade_frage_muster(topic: str, baustein: str) -> list[dict]:
-    """Vordefinierte Frage-Muster eines Bausteins aus dem Sidecar (leer = Fallback auf Live)."""
+async def lade_frage_muster(topic: str, baustein: str) -> list[dict]:
+    """Vordefinierte Frage-Muster eines Bausteins — DB-first, Fallback Sidecar (leer = Live)."""
+    rows = await db.list_frage_muster(topic, _norm_titel(baustein))
+    if rows:
+        return [{"subbaustein": r["sub_titel"], "typ": r["typ"], "frage": r["frage"]} for r in rows if r["frage"]]
     fm = _json_datei(frage_muster_path(topic))
     if not isinstance(fm, dict):
         return []
@@ -108,12 +124,24 @@ def lade_frage_muster(topic: str, baustein: str) -> list[dict]:
     ]
 
 
-def lade_uebersicht(topic: str) -> list[dict]:
-    """Strukturierte Baustein-Liste für die Übersicht: Titel + Beschreibung + Subbausteine/Stufen.
-
-    Verbindet bausteine.md (Nummer/Titel/Beschreibung) mit der Sidecar subbausteine.json
-    (Key = Titel). Fehlt die Sidecar, sind die Subbaustein-Listen leer.
-    """
+async def lade_uebersicht(topic: str) -> list[dict]:
+    """Strukturierte Baustein-Liste für die Übersicht — DB-first (Konsens + Subs/Stufen/Relevanz),
+    Fallback bausteine.md + Sidecar (Alt-Themen)."""
+    bs = await db.list_bausteine(topic, status="konsens")
+    if bs:
+        out = []
+        for num, b in enumerate(bs, 1):
+            subs = [s for s in await db.list_subbausteine(topic, b["titel_norm"]) if s["status"] == "konsens"]
+            out.append({
+                "num": num, "titel": b["titel"], "beschreibung": b["beschreibung"],
+                "subbausteine": [
+                    {"titel": s["sub_titel"],
+                     "stufe": s["stufe"] if s["stufe"] in _STUFEN else "mittel",
+                     "relevanz": s["relevanz"] if s["relevanz"] in ("relevant", "rand") else None}
+                    for s in subs if s["sub_titel"]
+                ],
+            })
+        return out
     entries = _lade_bausteine(_read(bausteine_path(topic)))
     sidecar = _json_datei(subbausteine_path(topic))
     sidecar = sidecar if isinstance(sidecar, dict) else {}
@@ -238,22 +266,14 @@ def cancel_bausteine(topic: str) -> bool:
 
 
 def _resume_step(topic: str) -> int:
-    """Erster noch offener Schritt anhand der persistierten Zwischendateien."""
+    """Erster noch offener Schritt anhand der persistierten Artefakte.
+    Inventar (Recherche→Konsolidierung→Klärung) gilt als fertig, sobald bausteine.md vorliegt."""
     files = _bausteine_files(topic)
     q = lade_quelle(topic)
     if q["type"] == "link" and not _crawl_fertig(topic):
         return _step_idx(topic, "Quelle laden")
-    if sum(p.exists() for p in files["recherche"]) < 3:
+    if not files["final"].exists():  # Inventar (DB-Loop) noch offen
         return _step_idx(topic, "Recherche")
-    if not files["recherche_mapping"].exists():
-        return _step_idx(topic, "Konsolidierung")
-    mapping = _mapping_schema(_json_datei(files["recherche_mapping"]))
-    geklaert = mapping is not None and (
-        not mapping[1]  # kein strittiger Rest
-        or any((r := _runde_schema(_json_datei(p))) is not None and not r[1] for p in files["mapping"].values())
-    )
-    if not geklaert:
-        return _step_idx(topic, "Klärung")
     if q["type"] == "projekt" and not files["ergaenzung"].exists():
         return _step_idx(topic, "Ergänzung")
     sidecar = _json_datei(files["sidecar"])
@@ -339,9 +359,10 @@ def _reset_ab_phase(topic: str, phase: int) -> None:
     if phase <= 2:  # Subbausteine
         files["sub_roh"].unlink(missing_ok=True)
         glob_del("subbaustein-*")
-    if phase <= 1:  # Inventar = kompletter Frischstart (alle Zwischendateien)
+    if phase <= 1:  # Inventar = kompletter Frischstart (alle Zwischendateien + bausteine.md)
         for p_alt in _alle_slot_dateien(files):
             p_alt.unlink(missing_ok=True)
+        files["final"].unlink(missing_ok=True)
 
 
 def _ergaenzung_schema(data):
@@ -386,14 +407,14 @@ def _pdfs_konvertieren(project: Path) -> None:
 _QUELLE_TEMPLATE = {"projekt": "Bausteine-Quelle-Projekt", "uni": "Bausteine-Quelle-Uni", "link": "Bausteine-Quelle-Link"}
 
 
-def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None) -> str:
+def _build_recherche_prompt(topic: str, out_path: Path, instructions: str, typ: str, ordner: Path | None, fokus: str = "") -> str:
     if typ in _QUELLE_TEMPLATE:
         source = _prompt(_QUELLE_TEMPLATE[typ], project=ordner)
     else:
         source = _prompt("Bausteine-Quelle-Thema", topic=topic)
     return _prompt(
         "Bausteine-Recherche",
-        topic=topic, source=source, bausteine_path=out_path, extra=_extra(instructions),
+        topic=topic, source=source, bausteine_path=out_path, fokus=fokus, extra=_extra(instructions),
     )
 
 
@@ -566,88 +587,143 @@ def _judge_block(chunk: list[int], entries: dict, daten: dict) -> str:
 
 
 async def _subbausteine_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict | None:
-    """Block B: drei Phasen mit Barriere — Finden, Wählen (Code-Merge), Klären.
-    Pro Phase laufen alle Pakete parallel; der Schritt bleibt, bis das letzte fertig ist.
-    → {Baustein-Titel: [Subbaustein, …]} oder None bei Abbruch/Fehler."""
+    """Block B (DB + Loop): je Paket Subbausteine in Runden finden (3 Finder, bis 0 neue/Kappe),
+    in der DB sammeln (≥2 Nennungen = Konsens, 1× verworfen), Judge bereinigt je Paket.
+    → {Baustein-Titel: [Subbaustein, …]} (Konsens) oder None. Befüllt DB-Tabelle `subbausteine`."""
     topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
     arbeit = files["arbeit"]
-    idx = _titel_index(entries)
     caps = "files" if quelle_ordner(topic) else "full"
     nums = list(entries)
     chunks = _chunk_nums(nums, _n_chunks(len(nums)))
     n = len(chunks)
+    titel_by_num = {num: _titel(entries[num]) for num in nums}
+    norm_by_num = {num: _norm_titel(titel_by_num[num]) for num in nums}
+    await db.delete_subbausteine(topic)  # Frischstart des Blocks (idempotenter Zähler)
 
-    def finder_paths(c):
-        return [arbeit / f"subbaustein-c{c}-{i}.md" for i in (1, 2, 3)]
+    async def _bekannt_block(chunk):
+        bl = []
+        for num in chunk:
+            subs = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num])]
+            if subs:
+                bl.append(f"<!-- baustein: {titel_by_num[num]} -->\n" + "\n".join(f"- {s}" for s in subs))
+        if not bl:
+            return ""
+        return "\n\nBEREITS GEFUNDEN — bestätige diese Subbausteine erneut UND ergänze fehlende:\n" + "\n".join(bl)
 
-    def final_path(c):
-        return arbeit / f"subbaustein-final-c{c}.md"
-
-    # Phase „Subbausteine finden": pro Paket 3 Finder (min. 2), alle Pakete parallel.
+    # Phase „Subbausteine finden": je Paket Loop bis 0 neue Subs / Zeit-Kappe.
     async def _finde(c, chunk):
-        paths = finder_paths(c)
-        vorhanden = sum(1 for p in paths if _parse_subbausteine(_read(p)))
-        if vorhanden >= 2:
-            return True
         zuteilung = "\n".join(f"- {entries[num]}" for num in chunk)
-        offen = [(i, p) for i, p in enumerate(paths, 1) if not _parse_subbausteine(_read(p))]
-        slots = [{
-            "key": f"bausteine-{topic}-subbaustein-c{c}-{i}",
-            "prompt": _prompt("Subbaustein-Recherche", topic=topic, zuteilung=zuteilung, out_path=p, extra=_extra(instructions)),
-            "role": "quick", "capabilities": caps,
-            "payload": (lambda result, p=p: _parse_subbausteine(_read(p)) or None),
-        } for i, p in offen]
-        neu = await _race(topic, f"Subbausteine Paket {c}", slots, 2 - vorhanden, _timeout("subbaustein", len(chunk)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE)
-        return not is_cancelled() and neu is not None
+        chunk_idx = _titel_index({num: titel_by_num[num] for num in chunk})
+        start = time.monotonic()
+        runde = 0
+        while not is_cancelled():
+            runde += 1
+            bekannt = await _bekannt_block(chunk) if runde > 1 else ""
+            paths = [arbeit / f"subbaustein-c{c}-r{runde}-{i}.md" for i in (1, 2, 3)]
+            for p in paths:
+                p.unlink(missing_ok=True)
+            slots = [{
+                "key": f"bausteine-{topic}-subbaustein-c{c}-r{runde}-{i}",
+                "prompt": _prompt("Subbaustein-Recherche", topic=topic, zuteilung=zuteilung, bekannt=bekannt, out_path=p, extra=_extra(instructions)),
+                "role": "quick", "capabilities": caps,
+                "payload": (lambda result, p=p: _parse_subbausteine(_read(p)) or None),
+            } for i, p in enumerate(paths, 1)]
+            texte = await _race(topic, f"Subbausteine Paket {c} R{runde}", slots, 2, _timeout("subbaustein", len(chunk)), provider, cancelled=is_cancelled, grace=KONSENS_GRACE)
+            if is_cancelled():
+                return False
+            if not texte:
+                return runde > 1  # Runde 1 ohne Ergebnis = Fehler; spätere = einfach Ende
+            vorhanden = {num: {s["sub_norm"] for s in await db.list_subbausteine(topic, norm_by_num[num])} for num in chunk}
+            neu = 0
+            for d in texte:
+                for marker, subs in d.items():
+                    num = _titel_aufloesen(chunk_idx, marker)
+                    if num is None:
+                        continue
+                    gesehen = set()
+                    for sub in subs:
+                        sn = _norm_titel(sub)
+                        if not sn or sn in gesehen:
+                            continue
+                        gesehen.add(sn)
+                        if sn not in vorhanden[num]:
+                            neu += 1
+                            vorhanden[num].add(sn)
+                        await db.upsert_subbaustein(topic, norm_by_num[num], sn, titel_by_num[num], sub)
+            if neu == 0:
+                break
+            if time.monotonic() - start > SUBBAUSTEIN_KAPPE:
+                _log(topic, f"Subbausteine Paket {c}: Zeit-Kappe erreicht (Runde {runde})")
+                break
+        return True
 
-    oks = await _gather_fortschritt([_finde(c, chunk) for c, chunk in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Subbausteine finden"))
+    oks = await _gather_fortschritt([_finde(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _melde_p(set_p, topic, "Subbausteine finden"))
     if is_cancelled():
         return None
     if not all(ok is True for ok in oks):
         _bausteine_errors[topic] = "Subbausteine fehlgeschlagen (Recherche)"
         return None
 
-    # Phase „Subbausteine wählen": Code-Merge je Paket (instant, kein Agent).
+    # Phase „Subbausteine wählen": ≥2 Nennungen = Konsens, 1× verworfen (Code).
     set_p(f"Subbausteine wählen ({n} Pakete)…", step=_step_idx(topic, "Subbausteine wählen"))
-    daten_by_c = {}
-    for c, chunk in enumerate(chunks, 1):
-        finder = [d for p in finder_paths(c) if (d := _parse_subbausteine(_read(p)))]
-        daten_by_c[c] = {num: _merge_finder(num, idx, finder) for num in chunk}
+    for num in nums:
+        for s in await db.list_subbausteine(topic, norm_by_num[num]):
+            await db.set_subbaustein_felder(topic, norm_by_num[num], s["sub_norm"],
+                                            status=("konsens" if s["nennungen"] >= 2 else "verworfen"))
 
-    # Phase „Subbausteine klären": Judge je Paket mit Strittigem, alle parallel.
+    # Phase „Subbausteine klären": Judge je Paket bereinigt die Konsens-Liste.
     async def _klaere(c, chunk):
-        daten = daten_by_c[c]
-        fp = final_path(c)
+        fp = arbeit / f"subbaustein-final-c{c}.md"
         if _parse_subbausteine(_read(fp)):
             return
-        if not any(daten[num][1] for num in chunk):
-            atomic_write_text(fp, _final_text(chunk, entries, daten))
+        bloecke, hat = [], False
+        for num in chunk:
+            subs = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num]) if s["status"] == "konsens"]
+            zeilen = "\n".join(f"- {s}" for s in subs) if subs else "- (keiner)"
+            bloecke.append(f"BAUSTEIN: {titel_by_num[num]}\nKonsens:\n{zeilen}")
+            if subs:
+                hat = True
+        if not hat:
             return
         status, _ = await run_single_slot(
             ctx, f"Subbaustein-Klärung {c}",
             key=f"bausteine-{topic}-subbaustein-final-c{c}",
-            prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine=_judge_block(chunk, entries, daten), out_path=fp, extra=_extra(instructions)),
+            prompt=_prompt("Subbaustein-Mapping", topic=topic, bausteine="\n\n".join(bloecke), out_path=fp, extra=_extra(instructions)),
             role="judge", capabilities="files",
             payload=lambda result, p=fp: _parse_subbausteine(_read(p)) or None,
             timeout=_timeout("subbaustein_check", len(chunk)),
         )
         if status == FAILED:
-            _log(topic, f"Subbaustein-Klärung Paket {c} fehlgeschlagen — nur Konsens übernommen")
+            _log(topic, f"Subbaustein-Klärung Paket {c} fehlgeschlagen — Konsens übernommen")
 
-    await _gather_fortschritt([_klaere(c, chunk) for c, chunk in enumerate(chunks, 1)], len(chunks), _melde_p(set_p, topic, "Subbausteine klären"))
+    await _gather_fortschritt([_klaere(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _melde_p(set_p, topic, "Subbausteine klären"))
     if is_cancelled():
         return None
 
-    # Fehlende finale Dateien → Konsens-Fallback; dann alle parsen.
+    # Finale Liste je Baustein: Judge-Ausgabe, sonst Konsens-Fallback. DB reconcilen + roh bauen.
     roh: dict[str, list[str]] = {}
     for c, chunk in enumerate(chunks, 1):
-        fp = final_path(c)
-        if not _parse_subbausteine(_read(fp)):
-            atomic_write_text(fp, _final_text(chunk, entries, daten_by_c[c]))
-        for marker, subs in (_parse_subbausteine(_read(fp)) or {}).items():
-            num = _titel_aufloesen(idx, marker)
-            if num is not None:
-                roh[_titel(entries[num])] = subs
+        final = _parse_subbausteine(_read(arbeit / f"subbaustein-final-c{c}.md")) or {}
+        chunk_idx = _titel_index({num: titel_by_num[num] for num in chunk})
+        final_by_num = {_titel_aufloesen(chunk_idx, m): subs for m, subs in final.items() if _titel_aufloesen(chunk_idx, m) is not None}
+        for num in chunk:
+            titel = titel_by_num[num]
+            konsens = [s["sub_titel"] for s in await db.list_subbausteine(topic, norm_by_num[num]) if s["status"] == "konsens"]
+            subs = final_by_num.get(num) or konsens
+            if not subs:
+                continue
+            roh[titel] = subs
+            # DB an die finale Liste angleichen: finale = konsens, Rest verworfen, Neues ergänzen.
+            final_norms = {_norm_titel(s) for s in subs}
+            have = {s["sub_norm"] for s in await db.list_subbausteine(topic, norm_by_num[num])}
+            for s in await db.list_subbausteine(topic, norm_by_num[num]):
+                await db.set_subbaustein_felder(topic, norm_by_num[num], s["sub_norm"],
+                                                status=("konsens" if s["sub_norm"] in final_norms else "verworfen"))
+            for s in subs:
+                sn = _norm_titel(s)
+                if sn and sn not in have:
+                    await db.upsert_subbaustein(topic, norm_by_num[num], sn, titel, s)
+                    await db.set_subbaustein_felder(topic, norm_by_num[num], sn, status="konsens")
     if not roh:
         _bausteine_errors[topic] = "Keine Subbausteine ermittelt"
         return None
@@ -1064,6 +1140,282 @@ async def _frage_muster_block(ctx: GenContext, set_p, files: dict, sidecar: dict
     return ergebnis
 
 
+# ── Inventar in der DB: Recherche-Loop · Konsolidierung · Klärung ────────────
+
+def _cited_sources(text: str) -> set[str]:
+    """Zitierte Quellen je Eintrag = 3. ' — '-Segment (URL bzw. Dateiname), kleingeschrieben."""
+    out = set()
+    for eintrag in _parse_auswahl(text).values():
+        teile = [t.strip() for t in eintrag.split(" — ")]
+        if len(teile) >= 3 and teile[-1]:
+            out.add(teile[-1].lower())
+    return out
+
+
+def _crawl_index(ordner) -> dict[str, str]:
+    """Alias (Dateiname ODER QUELLE:-URL, klein) → kanonischer Seiten-Key (Dateiname)."""
+    idx: dict[str, str] = {}
+    if not ordner or not Path(ordner).is_dir():
+        return idx
+    for p in sorted(Path(ordner).glob("*.txt")):
+        key = p.name
+        idx[key.lower()] = key
+        try:
+            erste = p.read_text(encoding="utf-8").splitlines()[0]
+        except (OSError, IndexError):
+            erste = ""
+        if erste.startswith("QUELLE:"):
+            url = erste[len("QUELLE:"):].strip()
+            if url:
+                idx[url.lower()] = key
+                idx[url.rstrip("/").lower()] = key
+    return idx
+
+
+def _abgedeckt(zitiert: set[str], crawl_idx: dict[str, str]) -> set[str]:
+    """Zitierte Quellen → Menge kanonischer Crawl-Seiten-Keys (was nicht passt, fällt weg)."""
+    out = set()
+    for z in zitiert:
+        key = crawl_idx.get(z) or crawl_idx.get(z.rstrip("/"))
+        if key:
+            out.add(key)
+    return out
+
+
+def _fokus_text(bereits: list[str], offen: list[str]) -> str:
+    """Re-Prompt-Block: noch nicht abgedeckte Crawl-Seiten + bereits gefundene Titel."""
+    teile = []
+    if offen:
+        liste = "\n".join(f"- {n}" for n in offen[:150])
+        teile.append(
+            "NOCH NICHT ABGEDECKTE QUELLEN-DATEIEN — lies ZUERST genau diese im Quell-Ordner "
+            f"und ergänze daraus die noch fehlenden Bausteine:\n{liste}"
+        )
+    if bereits:
+        liste = "\n".join(f"- {t}" for t in bereits)
+        teile.append("BEREITS GEFUNDEN (NICHT wiederholen — liefere nur NEUE Bausteine):\n" + liste)
+    return ("\n\n" + "\n\n".join(teile)) if teile else ""
+
+
+async def _set_inventar(topic: str, eintrag: str, status: str) -> None:
+    """Einen Inventar-Eintrag ('Titel — Beschreibung') mit Status in die DB schreiben."""
+    titel = _titel(eintrag)
+    norm = _norm_titel(titel)
+    if not norm:
+        return
+    teile = [t.strip() for t in eintrag.split(" — ")]
+    besch = teile[1] if len(teile) >= 2 else ""
+    await db.upsert_baustein(topic, norm, titel, besch)
+    await db.set_baustein_status(topic, norm, status)
+
+
+async def _recherche_loop(ctx: GenContext, set_p, files: dict, q: dict, ordner, instructions: str) -> bool:
+    """Füllt DB-Tabelle `bausteine` mit Kandidaten (+ Nennungszähler). Loop: je Runde
+    RECHERCHE_AGENTEN Agenten; re-promptet mit noch offenen Crawl-Seiten, bis alle gelesen /
+    0 neue / Zeit-Kappe. Resume: fertiger Schritt wird übersprungen. → True (ok) / False (Abbruch)."""
+    topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled
+    if await db.get_step_status(topic, "Recherche") == "fertig":
+        return True
+    arbeit = files["arbeit"]
+    caps = "files" if ordner else "full"
+    crawl_idx = _crawl_index(ordner)
+    alle_seiten = set(crawl_idx.values())
+
+    await db.delete_bausteine(topic)
+    await db.delete_coverage(topic)
+    await db.set_step_status(topic, "Recherche", "laufend")
+    set_p("Recherche läuft (Runde 1)…", step=_step_idx(topic, "Recherche"))
+
+    start = time.monotonic()
+    runde = 0
+    while not is_cancelled():
+        runde += 1
+        vorhandene = {b["titel_norm"] for b in await db.list_bausteine(topic)}
+        gelesen = set(await db.list_coverage(topic))
+        offen = sorted(alle_seiten - gelesen)
+        if runde > 1 and alle_seiten and not offen:
+            break  # volle Abdeckung erreicht
+        fokus = _fokus_text([b["titel"] for b in await db.list_bausteine(topic)], offen)
+        paths = [arbeit / f"recherche-r{runde}-{i}.md" for i in range(1, RECHERCHE_AGENTEN + 1)]
+        for p in paths:
+            p.unlink(missing_ok=True)
+        slots = [{
+            "key": f"bausteine-{topic}-recherche-r{runde}-{i}",
+            "prompt": _build_recherche_prompt(topic, p, instructions, q["type"], ordner, fokus=fokus),
+            "role": "quick", "capabilities": caps,
+            "payload": (lambda result, p=p: _file_payload(p)),
+        } for i, p in enumerate(paths, 1)]
+        set_p(f"Recherche läuft (Runde {runde})…")
+        texte = await _race(topic, f"Recherche R{runde}", slots, RECHERCHE_QUORUM,
+                            _timeout("recherche"), provider, cancelled=is_cancelled, grace=RECHERCHE_GRACE)
+        if is_cancelled():
+            return False
+        if not texte:
+            if runde == 1:
+                _bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)"
+                return False
+            break  # keine neuen Ergebnisse mehr
+        neu, zitiert = 0, set()
+        for text in texte:
+            zitiert |= _cited_sources(text)
+            gesehen = set()
+            for eintrag in _parse_auswahl(text).values():
+                titel = _titel(eintrag)
+                norm = _norm_titel(titel)
+                if not norm or norm in gesehen:
+                    continue
+                gesehen.add(norm)
+                teile = [t.strip() for t in eintrag.split(" — ")]
+                besch = teile[1] if len(teile) >= 2 else ""
+                quelle = [teile[2]] if len(teile) >= 3 and teile[2] else []
+                if norm not in vorhandene:
+                    neu += 1
+                    vorhandene.add(norm)
+                await db.upsert_baustein(topic, norm, titel, besch, quelle)
+        await db.mark_quellen_gelesen(topic, sorted(_abgedeckt(zitiert, crawl_idx)))
+        gesamt = len(await db.list_bausteine(topic))
+        deckung = len(await db.list_coverage(topic))
+        _log(topic, f"Recherche R{runde}: +{neu} neu (gesamt {gesamt}), Abdeckung {deckung}/{len(alle_seiten) or '?'}")
+        if neu == 0:
+            break
+        if time.monotonic() - start > RECHERCHE_KAPPE:
+            _log(topic, f"Recherche: Zeit-Kappe {RECHERCHE_KAPPE}s erreicht (Runde {runde})")
+            break
+    await db.set_step_status(topic, "Recherche", "fertig")
+    return True
+
+
+async def _konsolidiere(ctx: GenContext, set_p, files: dict) -> bool:
+    """Judge mergt Kandidaten semantisch + teilt in Konsens (≥2)/Rest (1×); Status in DB."""
+    topic, is_cancelled = ctx.topic, ctx.is_cancelled
+    if await db.get_step_status(topic, "Konsolidierung") == "fertig":
+        return True
+    set_p("Konsolidiere Recherche…", step=_step_idx(topic, "Konsolidierung"))
+    kandidaten = await db.list_bausteine(topic)
+    if not kandidaten:
+        _bausteine_errors[topic] = "Konsolidierung: keine Kandidaten"
+        return False
+    arbeit = files["arbeit"]
+    chunks = _chunk_nums(kandidaten, max(1, math.ceil(len(kandidaten) / KONSOLIDIERUNG_CHUNK)))
+    konsens, rest = [], []
+    for c, chunk in enumerate(chunks, 1):
+        fp = arbeit / f"konsolidierung-c{c}.json"
+        fp.unlink(missing_ok=True)
+        eintraege = "\n".join(
+            f"{i}. {b['titel']} — {b['beschreibung']} ({b['nennungen']}× genannt)" for i, b in enumerate(chunk, 1)
+        )
+        status, mapping = await run_single_slot(
+            ctx, f"Konsolidierung {c}",
+            key=f"bausteine-{topic}-konsolidierung-c{c}",
+            prompt=_prompt("Bausteine-Recherche-Mapping", topic=topic, n=RECHERCHE_AGENTEN, eintraege=eintraege, out_path=fp),
+            role="judge", capabilities="files",
+            payload=lambda result, p=fp: _mapping_schema(_json_datei(p)),
+            timeout=_timeout("recherche_mapping", len(chunk)),
+        )
+        if status == CANCELLED:
+            return False
+        if status == FAILED:
+            _bausteine_errors[topic] = "Recherche-Mapping fehlgeschlagen"
+            return False
+        k, r = mapping
+        konsens += k
+        rest += r
+    # Judge-Ausgabe ist maßgeblich → Inventar in der DB neu setzen.
+    await db.delete_bausteine(topic)
+    for t in konsens:
+        await _set_inventar(topic, t, "konsens")
+    for t in rest:
+        await _set_inventar(topic, t, "rest")
+    await db.set_step_status(topic, "Konsolidierung", "fertig")
+    return True
+
+
+async def _klaere_inventar(ctx: GenContext, set_p, files: dict) -> bool:
+    """1 Judge entscheidet über den Rest (1×-Genannte): aufnehmen → Konsens, sonst verworfen."""
+    topic, is_cancelled = ctx.topic, ctx.is_cancelled
+    if await db.get_step_status(topic, "Klärung") == "fertig":
+        return True
+    set_p("Klärung läuft…", step=_step_idx(topic, "Klärung"))
+    rest_rows = await db.list_bausteine(topic, status="rest")
+    if rest_rows:
+        konsens = [b["titel"] for b in await db.list_bausteine(topic, status="konsens")]
+        fp = files["arbeit"] / "klaerung.json"
+        fp.unlink(missing_ok=True)
+        status, ergebnis = await run_single_slot(
+            ctx, "Klärung",
+            key=f"bausteine-{topic}-klaerung",
+            prompt=_prompt(
+                "Bausteine-Klaerung", topic=topic,
+                konsens="\n".join(f"- {t}" for t in konsens) or "(noch leer)",
+                rest="\n".join(f"- {b['titel']}" for b in rest_rows),
+                final="\n- Entscheide JEDEN Eintrag. `rest` MUSS leer sein.",
+                out_path=fp,
+            ),
+            role="judge", capabilities="files",
+            payload=lambda result, p=fp: _runde_schema(_json_datei(p), final=True),
+            timeout=_timeout("auswahl_mapping", len(rest_rows)),
+        )
+        if status == CANCELLED:
+            return False
+        if status == FAILED:
+            _bausteine_errors[topic] = "Klärung fehlgeschlagen"
+            return False
+        aufnehmen, _ = ergebnis
+        auf_norm = {_norm_titel(_titel(t)) for t in aufnehmen}
+        for b in rest_rows:
+            await db.set_baustein_status(topic, b["titel_norm"], "konsens" if b["titel_norm"] in auf_norm else "verworfen")
+    await db.set_step_status(topic, "Klärung", "fertig")
+    return True
+
+
+async def _mirror_sidecar_db(topic: str, sidecar: dict) -> None:
+    """Sidecar {Baustein-Titel: [{titel, stufe, relevanz}]} in die DB-Tabelle subbausteine spiegeln."""
+    for btitel, subs in sidecar.items():
+        bnorm = _norm_titel(btitel)
+        if not bnorm or not isinstance(subs, list):
+            continue
+        for s in subs:
+            if not isinstance(s, dict):
+                continue
+            st = str(s.get("titel", "")).strip()
+            sn = _norm_titel(st)
+            if not sn:
+                continue
+            await db.put_subbaustein(topic, bnorm, sn, btitel, st,
+                                     stufe=s.get("stufe"), relevanz=s.get("relevanz"), status="konsens")
+
+
+async def _mirror_frage_muster_db(topic: str, muster: dict) -> None:
+    """Frage-Muster {Baustein-Titel: [{subbaustein, typ, frage}]} in die DB-Tabelle frage_muster spiegeln."""
+    await db.delete_frage_muster(topic)
+    for btitel, eintraege in muster.items():
+        bnorm = _norm_titel(btitel)
+        if not bnorm or not isinstance(eintraege, list):
+            continue
+        for e in eintraege:
+            if not isinstance(e, dict):
+                continue
+            sub = str(e.get("subbaustein", "")).strip()
+            sn = _norm_titel(sub)
+            typ = str(e.get("typ", "")).strip()
+            frage = str(e.get("frage", "")).strip()
+            if not (sn and typ and frage):
+                continue
+            await db.upsert_frage_muster(topic, bnorm, sn, btitel, sub, typ, frage)
+
+
+async def _reset_db_ab_phase(topic: str, ab_phase: int) -> None:
+    """DB-Inhalt der Phasen ≥ ab_phase verwerfen (Inventar=1 … Fragen=5)."""
+    if ab_phase <= 1:
+        await db.delete_bausteine(topic)
+        await db.delete_coverage(topic)
+        await db.delete_pipeline_state(topic, ["Recherche", "Konsolidierung", "Klärung"])
+    if ab_phase <= 2:
+        await db.delete_subbausteine(topic)
+    if ab_phase <= 5:
+        await db.delete_frage_muster(topic)
+
+
 async def generate_bausteine(topic: str, instructions: str = "", provider: str = DEFAULT_PROVIDER, ab_phase: int | None = None) -> None:
     if topic in _bausteine_progress:
         return
@@ -1096,6 +1448,7 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str =
             # unten wird übersprungen (er würde bei erhaltener Sidecar sonst alles wischen).
             if ab_phase is not None:
                 _reset_ab_phase(topic, ab_phase)
+                await _reset_db_ab_phase(topic, ab_phase)
             # Link-Quelle: erst crawlen (gleiche Domain, begrenzt) → wird zur Ordner-Quelle.
             if q["type"] == "link" and not _crawl_fertig(topic):
                 set_p("Quelle laden (Crawl)…", step=_step_idx(topic, "Quelle laden"))
@@ -1116,160 +1469,30 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str =
             if fertig:
                 for p_alt in _alle_slot_dateien(files):
                     p_alt.unlink(missing_ok=True)
+                await db.delete_pipeline_state(topic)
+                await db.delete_bausteine(topic)
+                await db.delete_subbausteine(topic)
+                await db.delete_frage_muster(topic)
+                await db.delete_coverage(topic)
 
-            # Schritt 1: 5 Recherche-Agenten, min. 3 mit Grace-Fenster — alle gültigen
-            # Slot-Dateien fließen ins Mapping (kein Kappen mehr bei 3)
-            recherchen: list[str] = []
-            offen = []
-            for i, path in enumerate(files["recherche"], 1):
-                text = _file_payload(path)
-                if text is not None:
-                    recherchen.append(text)
-                else:
-                    offen.append((i, path))
-            vorhanden = len(recherchen)
-            set_p(f"Recherche läuft ({vorhanden} gültig, min. 3)…", step=_step_idx(topic, "Recherche"))
-            if vorhanden < 3:
-                caps = "files" if ordner else "full"
-                slots = [
-                    {
-                        "key": f"bausteine-{topic}-recherche-{i}",
-                        "prompt": _build_recherche_prompt(topic, path, instructions, q["type"], ordner),
-                        "role": "quick", "capabilities": caps,
-                        "payload": (lambda result, p=path: _file_payload(p)),
-                    }
-                    for i, path in offen
-                ]
-                neue = await _race(
-                    topic, "Recherche", slots, 3 - vorhanden, _timeout("recherche"), provider,
-                    on_update=lambda c: set_p(f"Recherche läuft ({vorhanden + c} gültig, min. 3)…"),
-                    cancelled=is_cancelled, grace=RECHERCHE_GRACE,
-                )
+            # Inventar (DB): Recherche-Loop → Konsolidierung → Klärung.
+            if not await _recherche_loop(ctx, set_p, files, q, ordner, instructions):
                 if is_cancelled():
                     abgebrochen()
-                    return
-                if neue is None:
-                    _bausteine_errors[topic] = "Recherche fehlgeschlagen (Minimum nicht erreicht)"
-                    return
-                recherchen += neue
-
-            # Schritt 2: Recherche-Mapping — Code-Vormerge (exakte Titel) + 1 Agent
-            # für semantische Dubletten und Konsens/Rest-Teilung (fatal)
-            mapping = _mapping_schema(_json_datei(files["recherche_mapping"]))
-            if mapping is None:
-                set_p("Konsolidiere Recherche…", step=_step_idx(topic, "Konsolidierung"))
-                files["recherche_mapping"].unlink(missing_ok=True)
-                gemergt = _vormerge([_parse_auswahl(t) for t in recherchen])
-                eintraege = "\n".join(f"{i}. {text} ({n}× genannt)" for i, (text, n) in enumerate(gemergt, 1))
-                status, mapping = await run_single_slot(
-                    ctx, "Recherche-Mapping",
-                    key=f"bausteine-{topic}-recherche-mapping",
-                    prompt=_prompt(
-                        "Bausteine-Recherche-Mapping",
-                        topic=topic, n=len(recherchen), eintraege=eintraege,
-                        out_path=files["recherche_mapping"],
-                    ),
-                    role="judge", capabilities="files",
-                    payload=lambda result: _mapping_schema(_json_datei(files["recherche_mapping"])),
-                    timeout=_timeout("recherche_mapping", len(gemergt)),
-                )
-                if status == CANCELLED:
+                return
+            if not await _konsolidiere(ctx, set_p, files):
+                if is_cancelled():
                     abgebrochen()
-                    return
-                if status == FAILED:
-                    _bausteine_errors[topic] = "Recherche-Mapping fehlgeschlagen"
-                    return
-            konsens, rest = mapping
-
-            # Klärungs-Loop: 3 Auswahl-Agenten entscheiden über den Rest, ein
-            # Mapping-Agent sortiert in aufnehmen/verwerfen/weiter strittig.
-            # Leerer Rest beendet den Loop; Runde KONSENS_MAX_RUNDEN muss
-            # alles entscheiden. Der Konsens wächst nur hier im Code.
-            runde = 0
-            while rest and runde < KONSENS_MAX_RUNDEN:
-                runde += 1
-                final_runde = runde == KONSENS_MAX_RUNDEN
-                set_p(f"Klärung läuft (Runde {runde}/{KONSENS_MAX_RUNDEN})…", step=_step_idx(topic, "Klärung"))
-                mapping_path = files["mapping"][runde]
-
-                # Resume: fertiges Runden-Mapping wird direkt übernommen
-                ergebnis = _runde_schema(_json_datei(mapping_path), final=final_runde)
-                if ergebnis is None:
-                    mapping_path.unlink(missing_ok=True)
-                    konsens_block = "\n".join(f"- {t}" for t in konsens)
-                    rest_block = "\n".join(f"- {t}" for t in rest)
-
-                    # 3 Auswahl-Agenten, min. 2 mit Grace-Fenster
-                    entscheidungen = []
-                    offen = []
-                    for i, path in enumerate(files["auswahl"][runde], 1):
-                        res = _rest_schema(_json_datei(path))
-                        if res is not None:
-                            entscheidungen.append(res)
-                        else:
-                            offen.append((i, path))
-                    if len(entscheidungen) < 2:
-                        slots = [
-                            {
-                                "key": f"bausteine-{topic}-auswahl-r{runde}-{i}",
-                                "prompt": _prompt(
-                                    "Bausteine-Auswahl",
-                                    topic=topic, konsens=konsens_block, rest=rest_block, out_path=path,
-                                ),
-                                "role": "fast", "capabilities": "files",
-                                "payload": (lambda result, p=path: _rest_schema(_json_datei(p))),
-                            }
-                            for i, path in offen
-                        ]
-                        neue = await _race(
-                            topic, f"Auswahl r{runde}", slots, 2 - len(entscheidungen),
-                            _timeout("auswahl", len(rest)), provider,
-                            cancelled=is_cancelled, grace=KONSENS_GRACE,
-                        )
-                        if is_cancelled():
-                            abgebrochen()
-                            return
-                        if neue is None:
-                            _bausteine_errors[topic] = f"Auswahl fehlgeschlagen (Runde {runde}, Minimum nicht erreicht)"
-                            return
-                        entscheidungen += neue
-
-                    # Votum pro Rest-Eintrag deterministisch zählen
-                    indizes = [_titel_index(dict(enumerate(e, 1))) for e in entscheidungen]
-                    voten = "\n".join(
-                        f"{i}. {text} (von {sum(1 for idx in indizes if _titel_aufloesen(idx, text) is not None)}"
-                        f"/{len(entscheidungen)} Agenten übernommen)"
-                        for i, text in enumerate(rest, 1)
-                    )
-                    final_zusatz = (
-                        "\n- LETZTE RUNDE: Es gibt keine weitere Runde. `rest` MUSS leer sein"
-                        " — entscheide JEDEN Eintrag selbst: aufnehmen oder verwerfen."
-                        if final_runde else ""
-                    )
-                    status, ergebnis = await run_single_slot(
-                        ctx, f"Auswahl-Mapping r{runde}",
-                        key=f"bausteine-{topic}-auswahl-mapping-r{runde}",
-                        prompt=_prompt(
-                            "Bausteine-Auswahl-Mapping",
-                            topic=topic, n=len(entscheidungen), konsens=konsens_block,
-                            rest=voten, final=final_zusatz, out_path=mapping_path,
-                        ),
-                        role="judge", capabilities="files",
-                        payload=lambda result, p=mapping_path, f=final_runde: _runde_schema(_json_datei(p), final=f),
-                        timeout=_timeout("auswahl_mapping", len(rest)),
-                    )
-                    if status == CANCELLED:
-                        abgebrochen()
-                        return
-                    if status == FAILED:
-                        _bausteine_errors[topic] = f"Auswahl-Mapping fehlgeschlagen (Runde {runde})"
-                        return
-
-                aufnehmen, rest = ergebnis
-                _log(topic, f"Klärung Runde {runde}: {len(aufnehmen)} aufgenommen, {len(rest)} weiter strittig")
-                konsens = konsens + aufnehmen
-
-            entries = {i: t for i, t in enumerate(konsens, 1)}
+                return
+            if not await _klaere_inventar(ctx, set_p, files):
+                if is_cancelled():
+                    abgebrochen()
+                return
+            konsens_rows = await db.list_bausteine(topic, status="konsens")
+            entries = {
+                i: (f"{b['titel']} — {b['beschreibung']}" if b["beschreibung"] else b["titel"])
+                for i, b in enumerate(konsens_rows, 1)
+            }
 
             # Nur Projekte: Themenfeld-Ergänzung — Skript/Projekt ist ein Ausschnitt,
             # ein Web-Agent ergänzt kanonisch fehlende Bausteine, markiert mit [Ergänzung].
@@ -1360,6 +1583,14 @@ async def generate_bausteine(topic: str, instructions: str = "", provider: str =
                 if muster is None:
                     return  # Abbruch
                 atomic_write_json(files["frage_muster"], muster, indent=1)
+
+            # DB-Spiegel (Brücke): finalen Sidecar- + Frage-Muster-Stand in die DB schreiben.
+            sidecar = _json_datei(files["sidecar"])
+            if _sidecar_schema(sidecar) is not None:
+                await _mirror_sidecar_db(topic, sidecar)
+            muster = _json_datei(files["frage_muster"])
+            if isinstance(muster, dict) and muster:
+                await _mirror_frage_muster_db(topic, muster)
     except Exception as e:
         log.exception("[%s] Bausteine-Generierung fehlgeschlagen", topic)
         _bausteine_errors[topic] = str(e)[:2000]

backend/database.py

@@ -73,6 +73,102 @@ CREATE TABLE IF NOT EXISTS baustein_progress (
 )
 """
 
+# --- Bausteine-Pipeline-Inhalt (ersetzt Datei-Sidecars) ---
+
+# Inventar: ein Baustein je (topic, titel_norm). nennungen = Anzahl Agenten/Runden,
+# die ihn nannten (≥2 = Konsens). status: kandidat/konsens/rest/verworfen.
+CREATE_BAUSTEINE = """
+CREATE TABLE IF NOT EXISTS bausteine (
+    topic TEXT NOT NULL,
+    titel_norm TEXT NOT NULL,
+    titel TEXT NOT NULL,
+    beschreibung TEXT NOT NULL DEFAULT '',
+    nennungen INTEGER NOT NULL DEFAULT 1,
+    status TEXT NOT NULL DEFAULT 'kandidat',
+    quellen TEXT NOT NULL DEFAULT '[]',
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, titel_norm)
+)
+"""
+
+# Subbausteine je Baustein. stufe (einfach/mittel/schwer) + relevanz (relevant/rand)
+# werden später gesetzt. nennungen analog zum Inventar.
+CREATE_SUBBAUSTEINE = """
+CREATE TABLE IF NOT EXISTS subbausteine (
+    topic TEXT NOT NULL,
+    baustein_norm TEXT NOT NULL,
+    sub_norm TEXT NOT NULL,
+    baustein TEXT NOT NULL,
+    sub_titel TEXT NOT NULL,
+    nennungen INTEGER NOT NULL DEFAULT 1,
+    stufe TEXT,
+    relevanz TEXT,
+    status TEXT NOT NULL DEFAULT 'kandidat',
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, baustein_norm, sub_norm)
+)
+"""
+
+# Ein Frage-Muster je (Baustein, Subbaustein, Typ).
+CREATE_FRAGE_MUSTER = """
+CREATE TABLE IF NOT EXISTS frage_muster (
+    topic TEXT NOT NULL,
+    baustein_norm TEXT NOT NULL,
+    sub_norm TEXT NOT NULL,
+    baustein TEXT NOT NULL,
+    sub_titel TEXT NOT NULL,
+    typ TEXT NOT NULL,
+    frage TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, baustein_norm, sub_norm, typ)
+)
+"""
+
+# Crawl-Seiten-Abdeckung: treibt den Recherche-Loop (welche Quelle wurde zitiert?).
+CREATE_RECHERCHE_COVERAGE = """
+CREATE TABLE IF NOT EXISTS recherche_coverage (
+    topic TEXT NOT NULL,
+    quelle TEXT NOT NULL,
+    gelesen INTEGER NOT NULL DEFAULT 0,
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, quelle)
+)
+"""
+
+# Schritt-Status der Pipeline (ersetzt Datei-Existenz-Resume + Reset-Globs).
+CREATE_PIPELINE_STATE = """
+CREATE TABLE IF NOT EXISTS pipeline_state (
+    topic TEXT NOT NULL,
+    schritt TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'offen',
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, schritt)
+)
+"""
+
+# Fertiger Guide-Inhalt je (Thema, Format) als JSON-Blob (ersetzt die Guide-JSON-Datei).
+# Geteilt über alle Guide-Läufe desselben Thema+Formats (wie zuvor die Content-Datei).
+CREATE_GUIDE_CONTENT = """
+CREATE TABLE IF NOT EXISTS guide_content (
+    topic TEXT NOT NULL,
+    format TEXT NOT NULL,
+    json TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (topic, format)
+)
+"""
+
+# Quellen-Wahl je Thema (ersetzt quelle.json).
+CREATE_QUELLE = """
+CREATE TABLE IF NOT EXISTS quelle (
+    topic TEXT PRIMARY KEY,
+    type TEXT NOT NULL,
+    ort TEXT NOT NULL DEFAULT '',
+    spec TEXT NOT NULL DEFAULT '',
+    updated_at TEXT NOT NULL
+)
+"""
+
 _db: aiosqlite.Connection | None = None
 
 
@@ -95,6 +191,13 @@ async def init_db():
     await db.execute(CREATE_ELEMENTS)
     await db.execute(CREATE_BAUSTEIN_TEXTE)
     await db.execute(CREATE_BAUSTEIN_PROGRESS)
+    await db.execute(CREATE_BAUSTEINE)
+    await db.execute(CREATE_SUBBAUSTEINE)
+    await db.execute(CREATE_FRAGE_MUSTER)
+    await db.execute(CREATE_RECHERCHE_COVERAGE)
+    await db.execute(CREATE_PIPELINE_STATE)
+    await db.execute(CREATE_GUIDE_CONTENT)
+    await db.execute(CREATE_QUELLE)
     try:  # Migration für Bestands-DBs ohne step-Spalte
         await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER")
     except aiosqlite.OperationalError:
@@ -471,3 +574,269 @@ async def delete_baustein_daten(topic: str) -> None:
     await db.execute("DELETE FROM baustein_texte WHERE topic = ?", (topic,))
     await db.execute("DELETE FROM baustein_progress WHERE topic = ?", (topic,))
     await db.commit()
+
+
+# --- Bausteine-Pipeline-Inhalt: Inventar / Subbausteine / Frage-Muster / Coverage / State / Quelle ---
+
+async def upsert_baustein(topic: str, titel_norm: str, titel: str, beschreibung: str = "", quellen: list | None = None) -> None:
+    """Kandidat einfügen oder Nennungszähler erhöhen. Erst-Beschreibung bleibt erhalten."""
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO bausteine (topic, titel_norm, titel, beschreibung, nennungen, status, quellen, updated_at)
+           VALUES (?, ?, ?, ?, 1, 'kandidat', ?, ?)
+           ON CONFLICT(topic, titel_norm) DO UPDATE SET
+               nennungen = nennungen + 1, quellen = excluded.quellen, updated_at = excluded.updated_at""",
+        (topic, titel_norm, titel, beschreibung, json.dumps(quellen or [], ensure_ascii=False), _now()),
+    )
+    await db.commit()
+
+
+async def list_bausteine(topic: str, status: str | None = None) -> list[dict]:
+    db = await get_db()
+    if status is None:
+        cursor = await db.execute("SELECT * FROM bausteine WHERE topic = ? ORDER BY rowid", (topic,))
+    else:
+        cursor = await db.execute("SELECT * FROM bausteine WHERE topic = ? AND status = ? ORDER BY rowid", (topic, status))
+    rows = await cursor.fetchall()
+    out = []
+    for row in rows:
+        d = _row_to_dict(row, cursor)
+        d["quellen"] = json.loads(d.get("quellen") or "[]")
+        out.append(d)
+    return out
+
+
+async def set_baustein_status(topic: str, titel_norm: str, status: str, titel: str | None = None, beschreibung: str | None = None) -> None:
+    """Status setzen; optional Titel/Beschreibung aktualisieren (z.B. nach semantischem Merge)."""
+    db = await get_db()
+    fields = {"status": status, "updated_at": _now()}
+    if titel is not None:
+        fields["titel"] = titel
+    if beschreibung is not None:
+        fields["beschreibung"] = beschreibung
+    sets = ", ".join(f"{k} = :{k}" for k in fields)
+    await db.execute(
+        f"UPDATE bausteine SET {sets} WHERE topic = :topic AND titel_norm = :titel_norm",
+        {**fields, "topic": topic, "titel_norm": titel_norm},
+    )
+    await db.commit()
+
+
+async def delete_bausteine(topic: str) -> None:
+    db = await get_db()
+    await db.execute("DELETE FROM bausteine WHERE topic = ?", (topic,))
+    await db.commit()
+
+
+async def upsert_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str) -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, status, updated_at)
+           VALUES (?, ?, ?, ?, ?, 1, 'kandidat', ?)
+           ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET
+               nennungen = nennungen + 1, updated_at = excluded.updated_at""",
+        (topic, baustein_norm, sub_norm, baustein, sub_titel, _now()),
+    )
+    await db.commit()
+
+
+async def put_subbaustein(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str,
+                          stufe: str | None = None, relevanz: str | None = None, status: str = "konsens") -> None:
+    """Insert/Update OHNE Nennungszähler (Spiegel aus dem Sidecar). stufe/relevanz nur überschreiben,
+    wenn ein neuer Wert übergeben wird (COALESCE schützt Bestehendes)."""
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO subbausteine (topic, baustein_norm, sub_norm, baustein, sub_titel, nennungen, stufe, relevanz, status, updated_at)
+           VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?)
+           ON CONFLICT(topic, baustein_norm, sub_norm) DO UPDATE SET
+               baustein = excluded.baustein, sub_titel = excluded.sub_titel,
+               stufe = COALESCE(excluded.stufe, subbausteine.stufe),
+               relevanz = COALESCE(excluded.relevanz, subbausteine.relevanz),
+               status = excluded.status, updated_at = excluded.updated_at""",
+        (topic, baustein_norm, sub_norm, baustein, sub_titel, stufe, relevanz, status, _now()),
+    )
+    await db.commit()
+
+
+async def list_subbausteine(topic: str, baustein_norm: str | None = None) -> list[dict]:
+    db = await get_db()
+    if baustein_norm is None:
+        cursor = await db.execute("SELECT * FROM subbausteine WHERE topic = ? ORDER BY rowid", (topic,))
+    else:
+        cursor = await db.execute(
+            "SELECT * FROM subbausteine WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm)
+        )
+    rows = await cursor.fetchall()
+    return [_row_to_dict(row, cursor) for row in rows]
+
+
+async def set_subbaustein_felder(topic: str, baustein_norm: str, sub_norm: str, **fields) -> None:
+    """Setzt Felder (stufe/relevanz/status/sub_titel) einer Subbaustein-Zeile."""
+    fields["updated_at"] = _now()
+    sets = ", ".join(f"{k} = :{k}" for k in fields)
+    db = await get_db()
+    await db.execute(
+        f"UPDATE subbausteine SET {sets} WHERE topic = :topic AND baustein_norm = :baustein_norm AND sub_norm = :sub_norm",
+        {**fields, "topic": topic, "baustein_norm": baustein_norm, "sub_norm": sub_norm},
+    )
+    await db.commit()
+
+
+async def delete_subbausteine(topic: str) -> None:
+    db = await get_db()
+    await db.execute("DELETE FROM subbausteine WHERE topic = ?", (topic,))
+    await db.commit()
+
+
+async def upsert_frage_muster(topic: str, baustein_norm: str, sub_norm: str, baustein: str, sub_titel: str, typ: str, frage: str) -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO frage_muster (topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, updated_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+           ON CONFLICT(topic, baustein_norm, sub_norm, typ) DO UPDATE SET
+               frage = excluded.frage, updated_at = excluded.updated_at""",
+        (topic, baustein_norm, sub_norm, baustein, sub_titel, typ, frage, _now()),
+    )
+    await db.commit()
+
+
+async def list_frage_muster(topic: str, baustein_norm: str | None = None) -> list[dict]:
+    db = await get_db()
+    if baustein_norm is None:
+        cursor = await db.execute("SELECT * FROM frage_muster WHERE topic = ? ORDER BY rowid", (topic,))
+    else:
+        cursor = await db.execute(
+            "SELECT * FROM frage_muster WHERE topic = ? AND baustein_norm = ? ORDER BY rowid", (topic, baustein_norm)
+        )
+    rows = await cursor.fetchall()
+    return [_row_to_dict(row, cursor) for row in rows]
+
+
+async def delete_frage_muster(topic: str) -> None:
+    db = await get_db()
+    await db.execute("DELETE FROM frage_muster WHERE topic = ?", (topic,))
+    await db.commit()
+
+
+async def mark_quellen_gelesen(topic: str, quellen: list[str]) -> None:
+    """Markiert die zitierten Crawl-Seiten als gelesen (Recherche-Loop-Abdeckung)."""
+    if not quellen:
+        return
+    db = await get_db()
+    now = _now()
+    await db.executemany(
+        """INSERT INTO recherche_coverage (topic, quelle, gelesen, updated_at) VALUES (?, ?, 1, ?)
+           ON CONFLICT(topic, quelle) DO UPDATE SET gelesen = 1, updated_at = excluded.updated_at""",
+        [(topic, q, now) for q in quellen],
+    )
+    await db.commit()
+
+
+async def list_coverage(topic: str) -> dict[str, int]:
+    db = await get_db()
+    cursor = await db.execute("SELECT quelle, gelesen FROM recherche_coverage WHERE topic = ?", (topic,))
+    rows = await cursor.fetchall()
+    return {q: g for q, g in rows}
+
+
+async def delete_coverage(topic: str) -> None:
+    db = await get_db()
+    await db.execute("DELETE FROM recherche_coverage WHERE topic = ?", (topic,))
+    await db.commit()
+
+
+async def set_step_status(topic: str, schritt: str, status: str) -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO pipeline_state (topic, schritt, status, updated_at) VALUES (?, ?, ?, ?)
+           ON CONFLICT(topic, schritt) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at""",
+        (topic, schritt, status, _now()),
+    )
+    await db.commit()
+
+
+async def get_step_status(topic: str, schritt: str) -> str:
+    db = await get_db()
+    cursor = await db.execute(
+        "SELECT status FROM pipeline_state WHERE topic = ? AND schritt = ?", (topic, schritt)
+    )
+    row = await cursor.fetchone()
+    return row[0] if row else "offen"
+
+
+async def list_pipeline_state(topic: str) -> dict[str, str]:
+    db = await get_db()
+    cursor = await db.execute("SELECT schritt, status FROM pipeline_state WHERE topic = ?", (topic,))
+    rows = await cursor.fetchall()
+    return {s: st for s, st in rows}
+
+
+async def delete_pipeline_state(topic: str, schritte: list[str] | None = None) -> None:
+    db = await get_db()
+    if schritte is None:
+        await db.execute("DELETE FROM pipeline_state WHERE topic = ?", (topic,))
+    elif schritte:
+        marks = ",".join("?" for _ in schritte)
+        await db.execute(f"DELETE FROM pipeline_state WHERE topic = ? AND schritt IN ({marks})", (topic, *schritte))
+    await db.commit()
+
+
+async def set_quelle(topic: str, type: str, ort: str = "", spec: str = "") -> None:
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO quelle (topic, type, ort, spec, updated_at) VALUES (?, ?, ?, ?, ?)
+           ON CONFLICT(topic) DO UPDATE SET
+               type = excluded.type, ort = excluded.ort, spec = excluded.spec, updated_at = excluded.updated_at""",
+        (topic, type, ort, spec, _now()),
+    )
+    await db.commit()
+
+
+async def get_quelle(topic: str) -> dict | None:
+    db = await get_db()
+    cursor = await db.execute("SELECT type, ort, spec FROM quelle WHERE topic = ?", (topic,))
+    row = await cursor.fetchone()
+    if row is None:
+        return None
+    return {"type": row[0], "ort": row[1], "spec": row[2]}
+
+
+async def delete_quelle(topic: str) -> None:
+    db = await get_db()
+    await db.execute("DELETE FROM quelle WHERE topic = ?", (topic,))
+    await db.commit()
+
+
+async def set_guide_content(topic: str, format: str, content_json: str) -> None:
+    """Fertigen Guide-Inhalt (JSON-Blob) je Thema+Format speichern."""
+    db = await get_db()
+    await db.execute(
+        """INSERT INTO guide_content (topic, format, json, updated_at) VALUES (?, ?, ?, ?)
+           ON CONFLICT(topic, format) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""",
+        (topic, format, content_json, _now()),
+    )
+    await db.commit()
+
+
+async def get_guide_content(topic: str, format: str) -> str | None:
+    db = await get_db()
+    cursor = await db.execute("SELECT json FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
+    row = await cursor.fetchone()
+    return row[0] if row else None
+
+
+async def delete_guide_content(topic: str, format: str | None = None) -> None:
+    db = await get_db()
+    if format is None:
+        await db.execute("DELETE FROM guide_content WHERE topic = ?", (topic,))
+    else:
+        await db.execute("DELETE FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
+    await db.commit()
+
+
+async def delete_topic_pipeline(topic: str) -> None:
+    """Alle Pipeline-Inhalte eines Themas verwerfen (Inventar/Subs/Muster/Coverage/State/Quelle)."""
+    db = await get_db()
+    for tab in ("bausteine", "subbausteine", "frage_muster", "recherche_coverage", "pipeline_state", "quelle"):
+        await db.execute(f"DELETE FROM {tab} WHERE topic = ?", (topic,))
+    await db.commit()

backend/guide.py

@@ -9,6 +9,7 @@ Schritt-Dateien bleiben liegen → Abbruch erhält Fortschritt, ▶ setzt am off
 """
 
 import asyncio
+import json
 import logging
 import math
 from datetime import datetime, timezone
@@ -21,7 +22,7 @@ from config import (
     LESBARKEIT_AKTIV, TEMPLATES_DIR,
 )
 import lesbarkeit
-from database import list_guides, update_guide
+from database import list_guides, update_guide, list_bausteine, list_subbausteine, set_guide_content
 from fsutil import atomic_write_json, atomic_write_text
 from jsonio import read_json_file as _json_datei
 from paths import bausteine_path, guide_content_path, project_dir, subbausteine_path
@@ -52,12 +53,19 @@ GUIDE_CHUNK = 10
 LESE_RUNDEN = 1
 
 
-def _load_subbausteine(topic: str) -> dict[str, list[dict]]:
-    """Sidecar laden: {Baustein-Titel: [{titel, stufe}, …]}. Fehlt sie → {} (Fallback)."""
+async def _load_subbausteine(topic: str) -> dict[str, list[dict]]:
+    """Subbausteine je Baustein — DB-first ({titel, stufe, relevanz}), Fallback Sidecar-Datei.
+    Fehlt beides → {} (Guide nimmt alles)."""
+    out: dict[str, list[dict]] = {}
+    for r in await list_subbausteine(topic):
+        if r["status"] == "konsens" and r["sub_titel"] and r["stufe"] in ("einfach", "mittel", "schwer"):
+            out.setdefault(r["baustein"], []).append(
+                {"titel": r["sub_titel"], "stufe": r["stufe"], "relevanz": r["relevanz"]})
+    if out:
+        return out
     data = _json_datei(subbausteine_path(topic))
     if not isinstance(data, dict):
         return {}
-    out: dict[str, list[dict]] = {}
     for titel, subs in data.items():
         if not isinstance(subs, list):
             continue
@@ -478,9 +486,9 @@ async def _generate_sections(
         "Wähle, was diesem Zweck dient — lass weg, was dafür nicht nötig ist."
     )
 
-    # Subbausteine je Baustein (Sidecar) — früh geladen: steuert Auswahl + Sub-Filter je Format.
+    # Subbausteine je Baustein (DB-first) — früh geladen: steuert Auswahl + Sub-Filter je Format.
     # Fehlt sie → {} (Fallback: Guide nimmt alles).
-    subs_raw = _load_subbausteine(topic)
+    subs_raw = await _load_subbausteine(topic)
 
     def _hat_relevanz(num, art):
         return any(isinstance(s, dict) and s.get("relevanz") == art for s in subs_raw.get(_titel(entries[num]), []))
@@ -932,7 +940,13 @@ async def generate_guide(guide_id: str, topic: str, format_name: str, instructio
                 for p_alt in guide_slot_dateien(content_path):
                     p_alt.unlink(missing_ok=True)
 
-            alle = _lade_bausteine(bausteine_path(topic).read_text(encoding="utf-8"))
+            bs = await list_bausteine(topic, status="konsens")
+            if bs:
+                alle = {i: (f"{b['titel']} — {b['beschreibung']}" if b["beschreibung"] else b["titel"])
+                        for i, b in enumerate(bs, 1)}
+            else:  # Fallback: bausteine.md (Alt-Themen)
+                bp = bausteine_path(topic)
+                alle = _lade_bausteine(bp.read_text(encoding="utf-8")) if bp.exists() else {}
             if not alle:
                 await _fail(guide_id, "Keine Bausteine gefunden")
                 return
@@ -946,7 +960,8 @@ async def generate_guide(guide_id: str, topic: str, format_name: str, instructio
                 return
             content = {"topic": topic, "format": format_name, "chapters": chapters}
 
-            atomic_write_json(content_path, content, indent=1)
+            atomic_write_json(content_path, content, indent=1)  # Brücke (Resume/Fallback)
+            await set_guide_content(topic, format_name, json.dumps(content, ensure_ascii=False))
 
             now = datetime.now(timezone.utc).isoformat()
             await update_guide(guide_id, status="done", progress=None, step=None, updated_at=now)

backend/routes.py

@@ -4,7 +4,7 @@ import uuid
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, HTTPException
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, Response
 
 from agents import provider_available
 from config import PROJECTS_DIR, UNI_DIR, PROVIDERS
@@ -15,6 +15,7 @@ from database import (
     create_element, list_elements, get_element, update_element, delete_element,
     list_baustein_progress, get_baustein_progress, set_offene_frage,
     set_baustein_score, set_baustein_absolviert, set_baustein_verstanden, set_baustein_gemeistert, delete_baustein_daten,
+    delete_topic_pipeline, get_guide_content, delete_guide_content,
 )
 from bausteine import generate_bausteine, cancel_bausteine, bausteine_status, active_bausteine, reset_bausteine, lade_quelle, lade_uebersicht, subbausteine_titel, lade_frage_muster
 from elements import generate_element, chat_with_guide, chat_with_element, check_element, style_element, refine_suggestion
@@ -86,6 +87,8 @@ async def add_topic(req: TopicCreateRequest):
 async def remove_topic(topic: str):
     await delete_topic(topic)
     await delete_baustein_daten(topic)
+    await delete_topic_pipeline(topic)
+    await delete_guide_content(topic)
     shutil.rmtree(topic_dir(topic), ignore_errors=True)
     return {"ok": True}
 
@@ -170,6 +173,7 @@ async def cancel_bausteine_route(topic: str):
 @router.delete("/bausteine")
 async def remove_bausteine(topic: str):
     reset_bausteine(topic)
+    await delete_topic_pipeline(topic)
     return {"ok": True}
 
 
@@ -203,13 +207,13 @@ async def update_bausteine_quelle(req: BausteineQuelleUpdate):
 
 @router.get("/bausteine/uebersicht", response_model=list[BausteinUebersicht])
 async def get_bausteine_uebersicht(topic: str):
-    return lade_uebersicht(topic)
+    return await lade_uebersicht(topic)
 
 
 @router.get("/bausteine/frage-muster")
 async def get_frage_muster(topic: str, baustein: str):
     """Vordefinierte Frage-Muster eines Bausteins (leer = Fallback auf Live-Generierung)."""
-    return {"muster": lade_frage_muster(topic, baustein)}
+    return {"muster": await lade_frage_muster(topic, baustein)}
 
 
 # --- Baustein-Lernen: Chat, Prüfung ---
@@ -316,7 +320,7 @@ async def baustein_pruefung_route(req: BausteinPruefungRequest):
             frage = await pruefung_frage_variante(req.topic, req.baustein, req.section, kompakt, req.muster, provider=req.provider)
         else:
             # Fallback (kein Muster-Sidecar): Live-Generierung mit zufälligem Typ/Subbaustein.
-            subs = subbausteine_titel(req.topic, req.baustein)  # Fokus-Kandidaten (zufällig gewählt)
+            subs = await subbausteine_titel(req.topic, req.baustein)  # Fokus-Kandidaten (zufällig gewählt)
             frage = await pruefung_frage(req.topic, req.baustein, req.section, kompakt, msgs, subbausteine=subs, vermeide=req.vermeide, provider=req.provider)
         if frage is None:
             raise HTTPException(502, "Frage fehlgeschlagen — bitte erneut versuchen")
@@ -496,7 +500,10 @@ async def guide_content(guide_id: str):
         raise HTTPException(404, "Guide nicht gefunden")
     if guide["status"] != "done":
         raise HTTPException(404, "Inhalt nicht verfügbar")
-    path = guide_content_path(guide["topic"], guide["format"])
+    stored = await get_guide_content(guide["topic"], guide["format"])  # DB-first
+    if stored is not None:
+        return Response(content=stored, media_type="application/json")
+    path = guide_content_path(guide["topic"], guide["format"])  # Fallback: Datei (Alt-Themen)
     if not path.exists():
         raise HTTPException(404, "Datei nicht gefunden")
     return FileResponse(path, media_type="application/json")
@@ -611,6 +618,7 @@ async def remove(guide_id: str, slots: bool = False):
     # Content) bleibt fürs Resume erhalten, außer es wird explizit verlangt (slots=1).
     rest = [g for g in await list_guides() if g["topic"] == guide["topic"] and g["format"] == guide["format"]]
     if not rest:
+        await delete_guide_content(guide["topic"], guide["format"])
         content = guide_content_path(guide["topic"], guide["format"])
         if slots or content.exists():
             for p in guide_slot_dateien(content):

templates/Prompt/Bausteine-Klaerung.md

@@ -0,0 +1,17 @@
+Zum Thema "{topic}" sind unten Bausteine, die in der Recherche nur EINMAL genannt wurden. Die Quelle wurde vollständig durchsucht. Entscheide jeden Eintrag einzeln.
+
+BEREITS BESTÄTIGTES INVENTAR (nur Kontext — nicht ändern, nicht duplizieren):
+{konsens}
+
+EINZELN GENANNTE EINTRÄGE (jeweils entscheiden):
+{rest}
+
+Regeln:
+- Fachlich valider, eigenständiger Baustein und KEINE Dublette/Variante des Inventars → `aufnehmen`.
+- Erfunden, zu vage, oder Dublette/Variante eines Inventar-Bausteins → verwerfen (in keine Liste).
+- Übernimm aufgenommene Einträge WÖRTLICH ("Titel — Kurzbeschreibung"), nicht umformulieren.{final}
+
+Schreibe NUR die JSON-Datei nach: {out_path}
+
+Format (kein weiterer Text in der Datei):
+{{"aufnehmen": ["Titel — Kurzbeschreibung"], "rest": []}}

templates/Prompt/Bausteine-Recherche.md

@@ -15,4 +15,6 @@ Regeln:
 Schreibe NUR die Markdown-Datei nach: {bausteine_path}
 
 Format: GENAU eine Zeile pro Baustein: `N. Titel — Kurzbeschreibung — Quelle`
+Die Quelle (3. Segment) MUSS der exakte Dateiname bzw. die URL der Crawl-Seite sein, aus der der Baustein stammt — sie steuert die Abdeckungs-Prüfung.
+{fokus}
 {extra}

templates/Prompt/Subbaustein-Recherche.md

@@ -25,4 +25,5 @@ Schreibe NUR die Datei {out_path} — pro Baustein ein baustein-Marker (Titel EX
 - Zweiter Subbaustein
 
 Die Marker-Zeile exakt so schreiben. Kein Text außerhalb der Bausteine.
+{bekannt}
 {extra}