"""Guide board: schema parsers + card reset semantics (no LLM).""" import guide_board as gb TOPIC, FMT = "t", "Guide" def test_ziele_schema(): ok = gb._ziele_schema({"ziele": [{"id": "z1", "text": "Erklären, warum X", "sub": "S"}, {"id": "z2", "text": "Nennen von Y"}]}) assert [z["id"] for z in ok] == ["z1", "z2"] assert gb._ziele_schema({"ziele": []}) is None assert gb._ziele_schema({"ziele": [{"id": "z1", "text": "a"}, {"id": "z1", "text": "b"}]}) \ == [{"id": "z1", "text": "a", "sub": ""}] # duplicate ids fold assert gb._ziele_schema("quatsch") is None def test_gate_schema(): assert gb._gate_schema({"ok": True}) == [] claims = gb._gate_schema({"claims": [{"text": "Falsch", "grund": "fehlt"}]}) assert claims == [{"text": "Falsch", "grund": "fehlt", "urteil": "unbelegt"}] assert gb._gate_schema({}) is None # urteil "falsch" wird durchgereicht, alles andere defaultet auf unbelegt claims = gb._gate_schema({"claims": [{"text": "A", "grund": "widerspricht", "urteil": "FALSCH"}, {"text": "B", "grund": "x", "urteil": "quatsch"}]}) assert [c["urteil"] for c in claims] == ["falsch", "unbelegt"] # Voll-Inventur-Rauschen: als belegt begründete Einträge fliegen raus claims = gb._gate_schema({"claims": [{"text": "A", "grund": "Belegt durch Quelle X"}, {"text": "B", "grund": "nicht ableitbar"}]}) assert [c["text"] for c in claims] == ["B"] def test_pruefer_schema(): """Verschmolzenes Verdikt: ok-Kurzform, Claims-Normalisierung, Ziel-Vollständigkeit.""" assert gb._pruefer_schema({"ok": True}, {"z1"}) == { "claims": [], "ziele": {}, "luecken": [], "ballast": [], "lese_probleme": []} res = gb._pruefer_schema({"claims": [{"text": "A", "grund": "x", "urteil": "FALSCH"}], "ziele": {"z1": True, "z2": "false"}, "luecken": [{"ziel": "z2", "fehlt": "Beweis"}], "ballast": ["Abschweifung"], "lese_probleme": [{"problem": "zu lang"}]}, {"z1", "z2"}) assert res["claims"] == [{"text": "A", "grund": "x", "urteil": "falsch"}] assert res["ziele"] == {"z1": True, "z2": False} assert res["luecken"][0]["fehlt"] == "Beweis" and res["lese_probleme"] == ["zu lang"] assert gb._pruefer_schema({"ziele": {"z1": True}}, {"z1", "z2"}) is None # z2 fehlt assert gb._pruefer_schema({"lese_probleme": []}, set()) is not None # leeres Verdikt ok assert gb._pruefer_schema("quatsch", set()) is None def test_auftraege_schwelle_und_kritisch(): """1–2 nur-unbelegt-Claims verfallen (GATE_FIX_MIN); falsch/Lücken sind kritisch.""" leer = {"claims": [], "ziele": {}, "luecken": [], "ballast": [], "lese_probleme": []} z, k = gb._auftraege({**leer, "claims": [{"text": "c", "grund": "", "urteil": "unbelegt"}] * 2}, []) assert z == [] and k is False z, k = gb._auftraege({**leer, "claims": [{"text": "c", "grund": "w", "urteil": "falsch"}]}, []) assert len(z) == 1 and k is True z, k = gb._auftraege({**leer, "luecken": [{"ziel": "z1", "fehlt": "X"}]}, ["Länge 2000"]) assert len(z) == 2 and k is True async def test_reset_from_stage(testdb): db = testdb await db.upsert_guide_card(TOPIC, FMT, "a", "A") await db.upsert_guide_card(TOPIC, FMT, "b", "B") await db.set_guide_card(TOPIC, FMT, "a", stage="done", md="text", writer_rounds=2) await db.set_guide_card(TOPIC, FMT, "b", stage="pruefer", md="text") await db.put_lernziel(TOPIC, "a", "z1", "Ziel") # reset ab writer (idx 2): beide Karten zurück, md geleert, Ziele bleiben moved = await gb.reset_from_stage(TOPIC, FMT, 2) assert moved == 2 cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, FMT)} assert cards["a"]["stage"] == "writer" and cards["a"]["md"] == "" and cards["a"]["writer_rounds"] == 0 assert cards["b"]["stage"] == "writer" assert await db.list_lernziele(TOPIC, "a") # reset ab lernziele (idx 0): Ziele weg await gb.reset_from_stage(TOPIC, FMT, 0) assert not await db.list_lernziele(TOPIC, "a") assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "lernziele" async def test_done_step(testdb): db = testdb assert await gb.done_step(TOPIC, FMT) == -1 await db.upsert_guide_card(TOPIC, FMT, "a", "A") assert await gb.done_step(TOPIC, FMT) == -1 # alles in lernziele await db.set_guide_card(TOPIC, FMT, "a", stage="pruefer") assert await gb.done_step(TOPIC, FMT) == 2 # bis writer fertig await db.set_guide_card(TOPIC, FMT, "a", stage="done") assert await gb.done_step(TOPIC, FMT) == len(gb.GUIDE_STAGES) async def test_run_card_sets_and_clears_live_info(testdb, monkeypatch): """Regression: _live nutzte env.format_name (existiert nicht) → AttributeError beim ersten Stage-Start. Treibt eine Karte durch _run_card mit Fake-Stage.""" import asyncio from types import SimpleNamespace import guide_board as gb db = testdb await db.upsert_guide_card("t", "Guide", "alpha", "Alpha") env = SimpleNamespace(ctx=None, guide_id="g-live", topic="t", format="Guide") card = {"block_norm": "alpha", "block": "Alpha", "stage": "lernziele", "status": "open"} seen = {} async def fake_stage(env2, card2): seen.update(dict(gb._live_info)) card2["stage"] = "done" return True monkeypatch.setattr(gb, "_STAGE_FN", {"lernziele": fake_stage}) await gb._run_card(env, card, asyncio.Semaphore(1)) assert card["stage"] == "done" assert ("t", "Guide", "alpha") in seen # live info stand während der Stage assert ("t", "Guide", "alpha") not in gb._live_info # und wurde aufgeräumt def test_merge_split_sections_one_section_all_markers(): import guide_board as gb from textkit import _parse_fragment a = _parse_fragment(""" Kurzer Einstieg kompakt. YAML kompakt. Einstieg ausführlich. YAML ausführlich.""")[0] b = _parse_fragment(""" TOML kompakt. Unerwünschter zweiter Einstieg. TOML ausführlich.""")[0] merged = gb._merge_split_sections(a, b) secs = _parse_fragment(merged) assert len(secs) == 1 sec = secs[0] assert sec["title"] == "Front Matter" assert [s["title"] for s in sec["subs"]] == ["YAML-Basics", "TOML-Sektionen"] assert sec["anchor"] == "Einstieg ausführlich." # Teil-B-Einstieg verworfen assert "TOML ausführlich." in sec["md"] and "YAML kompakt." in sec["compact"] async def test_card_examples_filters_and_formats(testdb): """_card_examples: Norm-Matching auf die übergebenen Subs; unmatchte Beispiele nur beim Voll-Writer/Teil 1 (include_unmatched) — nie stillschweigend weg.""" import json as _json import guide_board as gb from types import SimpleNamespace db = testdb await db.put_sub_artifact("t", "gross", "sub eins", "example", _json.dumps({"problem": "P1", "steps": ["a", "b"], "result": "R1"}), "Gross", "Sub Eins") await db.put_sub_artifact("t", "gross", "verwaist", "example", _json.dumps({"problem": "P2", "steps": ["x"], "result": "R2"}), "Gross", "Verwaister Sub") env = SimpleNamespace(topic="t") subs = [{"title": "Sub Eins", "level": "beginner"}] full = await gb._card_examples(env, "gross", subs) assert "Sub Eins" in full and "P1" in full and "1) a 2) b" in full and "R1" in full assert "Subbaustein unklar" in full and "P2" in full # orphan attached with hint half = await gb._card_examples(env, "gross", subs, include_unmatched=False) assert "P1" in half and "P2" not in half # split half: only its own subs assert await gb._card_examples(env, "leer", subs) == "" def test_writer_template_has_examples_placeholder(): """Smoke: alle Platzhalter versorgt — ein fehlender Kwarg stürbe als KeyError.""" from pipeline import _prompt text = _prompt("Guide-Writer-Board", topic="t", format_name="Guide", chapter="K1", assignment="- B", ziele="- z", facts="F", examples="", gaps="", budget=2000, spec="", out_path="/tmp/x.md", extra="") assert "VERIFIED FACTS" in text and "2000 characters" in text async def test_pruefer_counts_examples_as_facts(testdb, monkeypatch, tmp_path): """Prüfer-Prompt enthält die Beispiele als verifizierte Fakten — sonst fliegen gerechnete Beispielwerte als „nicht belegt" raus.""" import json as _json import guide_board as gb from types import SimpleNamespace db = testdb await db.upsert_guide_card("t", "Guide", "gross", "Gross") await db.put_sub_artifact("t", "gross", "sub eins", "example", _json.dumps({"problem": "P1", "steps": ["a"], "result": "R1"}), "Gross", "Sub Eins") captured = {} async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout, on_line=None): captured["prompt"] = prompt return "ok", payload((0, '{"ok": true}', "")) monkeypatch.setattr(gb, "run_single_slot", fake_slot) monkeypatch.setattr(gb, "_card_facts", lambda e, b: "FAKT X") monkeypatch.setattr(gb, "READABILITY_ACTIVE", False) env = SimpleNamespace(ctx=SimpleNamespace(topic="t", provider="p", is_cancelled=lambda: False), guide_id="g", topic="t", format="Guide", instructions="", subs_by_title={"Gross": [{"title": "Sub Eins", "level": "beginner"}]}, spec="", slot=lambda name: tmp_path / name) md = ("\n\n" + "Text im Rahmen. " * 20) card = {"block_norm": "gross", "block": "Gross", "stage": "pruefer", "status": "open", "writer_rounds": 0, "gate_info": "", "md": md} ok = await gb._stage_pruefer(env, card) assert ok is True assert "VERIFIED WORKED EXAMPLES" in captured["prompt"] and "P1" in captured["prompt"] async def test_writer_splits_oversized_first_draft(testdb, monkeypatch, tmp_path): import guide_board as gb from types import SimpleNamespace db = testdb await db.upsert_guide_card("t", "Guide", "gross", "Gross") calls = [] async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout, on_line=None): calls.append(label) part = "2" if key.endswith("-b") else "1" p = tmp_path / f"out-{key[-1]}.md" p.write_text(f"\n\n" + ("Einstieg.\n" if part == "1" else "") + f"\nText {part}.", encoding="utf-8") # payload liest die ECHTE Slot-Datei — wir schreiben direkt an deren Pfad import re as _re m = _re.search(r"(/\S+\.md)", prompt) with open(m.group(1), "w", encoding="utf-8") as f: f.write(p.read_text(encoding="utf-8")) return "ok", payload(None) monkeypatch.setattr(gb, "run_single_slot", fake_slot) subs = [{"title": f"Sub {i}", "level": "beginner", "relevance": "relevant"} for i in range(31)] env = SimpleNamespace(ctx=SimpleNamespace(topic="t", provider="p", is_cancelled=lambda: False), guide_id="g", topic="t", format="Guide", instructions="", subs_by_title={"Gross": subs}, spec="", slot=lambda name: tmp_path / name) monkeypatch.setattr(gb, "_card_facts", lambda e, b: "") card = {"block_norm": "gross", "block": "Gross", "stage": "writer", "status": "open", "writer_rounds": 0, "gate_info": "", "md": "", "chapter": "K1"} ok = await gb._stage_writer(env, card) assert ok is True assert [c for c in calls if "(1/2)" in c] and [c for c in calls if "(2/2)" in c] from textkit import _parse_fragment secs = _parse_fragment(card["md"]) assert len(secs) == 1 and [s["title"] for s in secs[0]["subs"]] == ["Sub 1", "Sub 2"] assert card["stage"] == "pruefer" async def test_lernziele_retry_bei_leerer_liste(testdb, tmp_path, monkeypatch): """Leere Ziele-Liste → genau EIN Ersatz-Versuch (Key-Suffix -2); dessen Ziele landen in der DB.""" db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-r", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": [{"title": "S1", "level": "beginner"}]}, {}, "(quelle)", "spec") card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0} calls = [] async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): calls.append(key) if len(calls) == 1: return gb.OK, [] return gb.OK, [{"id": "z1", "text": "Ziel", "sub": "S1"}] monkeypatch.setattr(gb, "run_single_slot", fake_slot) assert await gb._stage_lernziele(env, card) assert len(calls) == 2 and calls[1].endswith("-2") assert [z["ziel_id"] for z in await db.list_lernziele(TOPIC, "alpha")] == ["z1"] async def test_lernziele_zweimal_leer_laeuft_weiter(testdb, tmp_path, monkeypatch): db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-r2", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": []}, {}, "(q)", "spec") card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0} async def leer(ctx, label, *, key, prompt, role, capabilities, payload, timeout): return gb.OK, [] monkeypatch.setattr(gb, "run_single_slot", leer) assert await gb._stage_lernziele(env, card) assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "zuweisung" assert not await db.list_lernziele(TOPIC, "alpha") async def test_pruefer_text_sink_ohne_befund_done(testdb, tmp_path, monkeypatch): """Prüfer antwortet als Text (Engine-Sink, capabilities none); ohne Befund → done.""" db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-l", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": []}, {}, "(q)", "spec") md = ("\n\n\n- x\n" "\n" + "Text im Längen-Rahmen. " * 20) # ~460 Z. — kein Längen-Trigger card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0, "md": md} seen = {} async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): seen["caps"] = capabilities return gb.OK, payload((0, '{"ok": true}', "")) monkeypatch.setattr(gb, "run_single_slot", fake_slot) monkeypatch.setattr(gb, "READABILITY_ACTIVE", False) assert await gb._stage_pruefer(env, card) assert seen["caps"] == "none" assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "done" async def test_writer_prompt_traegt_budget(testdb, tmp_path, monkeypatch): db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-w", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": [{"title": "S1", "level": "beginner"}, {"title": "S2", "level": "beginner"}]}, {}, "(q)", "spec") card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0, "chapter": "K", "gate_info": ""} seen = {} async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): seen["prompt"] = prompt return gb.FAILED, None monkeypatch.setattr(gb, "run_single_slot", fake_slot) await gb._stage_writer(env, card) assert str(gb.block_budget(env.subs_by_title["Alpha"])) in seen["prompt"] # 2× Basis-Budget async def test_pruefer_claims_schwelle(testdb, tmp_path, monkeypatch): """1–2 nur-unbelegt-Claims → kein Fix (Karte direkt done); die Schwelle lebt in _auftraege.""" db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-g", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": []}, {}, "(q)", "spec") md = "\n\n" + "Text im Rahmen. " * 20 card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0, "md": md} async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): antwort = '{"claims": [{"text": "c1", "grund": ""}, {"text": "c2", "grund": ""}]}' return gb.OK, payload((0, antwort, "")) monkeypatch.setattr(gb, "run_single_slot", fake_slot) monkeypatch.setattr(gb, "READABILITY_ACTIVE", False) assert await gb._stage_pruefer(env, card) assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "done" async def test_load_subblocks_defaultet_levellose(testdb): """Consensus-Row ohne level fällt NICHT mehr raus — Default 'advanced'. (Re-Run-Resume hinterließ 25 solcher Rows; der Writer verlor sie stumm.)""" from guide import _load_subblocks db = testdb await db.put_subblock("t", "block", "mit level", "Block", "Mit Level", level="beginner", status="consensus") await db.put_subblock("t", "block", "ohne level", "Block", "Ohne Level", status="consensus") subs = await _load_subblocks("t") by_title = {s["title"]: s["level"] for s in subs["Block"]} assert by_title == {"Mit Level": "beginner", "Ohne Level": "advanced"} async def test_falsch_claim_erzwingt_fix_und_repruefer(testdb, tmp_path, monkeypatch): """Ein einzelner FALSCH-Claim läuft in den Fix, auch unter GATE_FIX_MIN; nach angewandtem Fix läuft GENAU EIN Re-Prüfer-Pass (der alte Lese-Fix blieb ungeprüft).""" db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-gf", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": []}, {}, "(q)", "spec") md = "\n\n" + "Text im Rahmen. " * 20 card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0, "md": md, "gate_info": ""} keys = [] async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): keys.append(key) if "-pruef-" in key and key.endswith("-re"): return gb.OK, payload((0, '{"ok": true}', "")) if "-pruef-" in key: return gb.OK, payload((0, '{"claims": [{"text": "c1", "grund": "widerspricht", "urteil": "falsch"}]}', "")) if "-gfix-" in key: # Fix liefert eine valide Section über die Slot-Datei import re as _re m = _re.search(r"(/\S+\.md)", prompt) with open(m.group(1), "w", encoding="utf-8") as f: f.write(md.replace("Text im Rahmen.", "Korrigiert.")) return gb.OK, payload(None) raise AssertionError(key) monkeypatch.setattr(gb, "run_single_slot", fake_slot) monkeypatch.setattr(gb, "READABILITY_ACTIVE", False) assert await gb._stage_pruefer(env, card) karte = (await db.list_guide_cards(TOPIC, FMT))[0] assert karte["stage"] == "fix" and karte["gate_info"].startswith("KRITISCH") card.update(stage="fix", gate_info=karte["gate_info"]) assert await gb._stage_fix(env, card) assert any("-gfix-" in k for k in keys) assert any(k.endswith("-re") for k in keys) # Re-Prüfer lief karte = (await db.list_guide_cards(TOPIC, FMT))[0] assert karte["stage"] == "done" and "Korrigiert." in karte["md"] async def test_laengen_trigger_startet_fix(testdb, tmp_path, monkeypatch): """Ausführlich-Teil über der Obergrenze → deterministisches Längen-Problem mit hartem Zeichenziel landet als Auftrag in der Fix-Stage (unkritisch → kein Re-Prüfer).""" db = testdb await db.upsert_guide_card(TOPIC, FMT, "alpha", "Alpha") env = gb._Env(None, "g-lz", TOPIC, FMT, "", tmp_path / "Guide.json", {"Alpha": [{"title": "S1", "level": "beginner", "relevance": "relevant"}]}, {}, "(q)", "spec") md = ("\n\n- x\n\n" + "Viel zu langer Sockeltext. " * 80) # ~2160 Z./Sub > 1200×0.9 card = {"block_norm": "alpha", "block": "Alpha", "writer_rounds": 0, "md": md, "gate_info": ""} seen = {} async def fake_slot(ctx, label, *, key, prompt, role, capabilities, payload, timeout): if "-gfix-" in key: seen["auftraege"] = prompt return gb.FAILED, None if key.endswith("-re"): raise AssertionError("unkritischer Befund darf keinen Re-Prüfer starten") return gb.OK, payload((0, '{"ok": true}', "")) # Prüfer: keine LLM-Befunde monkeypatch.setattr(gb, "run_single_slot", fake_slot) monkeypatch.setattr(gb, "READABILITY_ACTIVE", False) assert await gb._stage_pruefer(env, card) assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "fix" card.update(stage="fix", gate_info=(await db.list_guide_cards(TOPIC, FMT))[0]["gate_info"]) assert await gb._stage_fix(env, card) budget = gb.block_budget(env.subs_by_title["Alpha"]) assert "Länge" in seen["auftraege"] and f"etwa {budget} Zeichen GESAMT" in seen["auftraege"] async def test_repair_karten_setzt_befundkarten_auf_pruefer(testdb, tmp_path, monkeypatch): """Guide-Repair: Karten mit QA-Befunden (alle Befundklassen-Formate) → pruefer, md bleibt; Karten ohne Befund unangetastet. Kein Report → leere Liste.""" import json as _json import qa as qa_mod db = testdb monkeypatch.setattr(qa_mod, "QA_DIR", tmp_path / "qa") for n in ("alpha", "beta", "gamma"): await db.upsert_guide_card(TOPIC, FMT, n, n.title()) await db.set_guide_card(TOPIC, FMT, n, stage="done", status="ok", md="\nText") assert await gb.repair_karten(TOPIC, FMT) == [] # kein Report tdir = tmp_path / "qa" / TOPIC tdir.mkdir(parents=True) (tdir / "guide-20260705-000000.json").write_text(_json.dumps({ "marker_fehlend": ["Alpha · sub eins"], "fachlich_falsch": ["Beta"], "ziel_ohne_anker": [], "laengen_ausreisser": [], "redundanz": [], "lesbarkeit": [], }), encoding="utf-8") betroffen = await gb.repair_karten(TOPIC, FMT) assert sorted(betroffen) == ["Alpha", "Beta"] cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, FMT)} assert cards["alpha"]["stage"] == cards["beta"]["stage"] == "pruefer" assert cards["alpha"]["md"] # Text bleibt — der Prüfer arbeitet auf dem Bestand assert cards["gamma"]["stage"] == "done"