marek/creator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Team3
2026-07-05 00:25:53 +02:00
parent 4105146c59
commit d25824229e
15 changed files with 392 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

370
backend/guide_board.py
View File

@@ -3,9 +3,10 @@
lernziele judge-Rolle Backward Design — objectives BEFORE writing
zuweisung code chapter/order from the outline artefact + facts grounding
writer guide-Rolle ONE coherent per-block text, only from VERIFIED FACTS
fakten_gate judge-Rolle CoVe: atomic claims, each binary against the facts → minimal fix
coverage judge-Rolle objective↔section mapping; gap → back to writer (max 2 rounds)
lesbarkeit judge-Rolle Lese-Check + deterministic readability gate → fix → done
pruefer judge-Rolle EIN Call: CoVe-Fakten + Coverage + Lesbarkeit (lasen vorher
denselben Text in 3 seriellen Calls) + deterministische Gates
fix guide-Rolle EIN Rewrite unter allen Auflagen; bei falsch/Lücken danach
genau ein Re-Prüfer-Pass (der alte Lese-Fix blieb ungeprüft)
Runner: one asyncio task per card (cards are fixed from the start — no queue engine
needed); stage transitions are persisted in guide_cards, so the board is live and
@@ -31,11 +32,10 @@ from textkit import _norm_title, _parse_fragment, _title
log = logging.getLogger("creator.guide_board")
GUIDE_STAGES = ("lernziele", "zuweisung", "writer", "fakten_gate", "coverage", "lesbarkeit")
GUIDE_STAGES = ("lernziele", "zuweisung", "writer", "pruefer", "fix")
STAGE_LABELS = {"lernziele": "Lernziele", "zuweisung": "Zuweisung", "writer": "Writer",
"fakten_gate": "Fakten-Gate", "coverage": "Coverage",
"lesbarkeit": "Lesbarkeit", "done": "Fertig"}
from config import GATE_FIX_MIN, MAX_WRITER_ROUNDS, WRITER_SPLIT_SUBS # zentral tunebar
"pruefer": "Prüfen", "fix": "Fix", "done": "Fertig"}
from config import GATE_FIX_MIN, WRITER_SPLIT_SUBS # zentral tunebar
# Simultaneous cards = the per-topic agent cap: every card busies exactly ONE agent at a
# time (its stages run serially), so a lower number just idles slots (was hardcoded 10
# from the old 10-slot era while the .env already allowed 24).
@@ -87,33 +87,31 @@ def _gate_schema(data):
return out
def _coverage_schema(data, ziel_ids: set[str]):
"""{"ziele":{id:bool}, "luecken":[{ziel,fehlt}], "ballast":[str]} — ziele must cover all ids."""
if not isinstance(data, dict) or not isinstance(data.get("ziele"), dict):
def _pruefer_schema(data, ziel_ids: set[str]):
"""Verschmolzenes Prüfer-Verdikt: Claims (Fakten-Gate-Semantik via _gate_schema) +
Coverage (ziele/luecken/ballast) + Lesbarkeit (lese_probleme). {"ok":true} = leeres
Verdikt. `ziele` muss alle ids abdecken, wenn Ziele existieren — sonst optional."""
if not isinstance(data, dict):
return None
if data.get("ok") is True:
return {"claims": [], "ziele": {}, "luecken": [], "ballast": [], "lese_probleme": []}
if not any(k in data for k in ("claims", "ziele", "luecken", "ballast", "lese_probleme")):
return None
claims = _gate_schema({"claims": data["claims"]}) if data.get("claims") else []
if claims is None:
return None
ziele = {}
for k, v in data["ziele"].items():
for k, v in (data.get("ziele") or {}).items() if isinstance(data.get("ziele"), dict) else []:
ziele[str(k)] = str(v).strip().casefold() in ("true", "ja", "yes", "1")
if not ziel_ids <= set(ziele):
if ziel_ids and not ziel_ids <= set(ziele):
return None
luecken = [{"ziel": str(l.get("ziel", "")), "fehlt": str(l.get("fehlt", ""))}
for l in data.get("luecken", []) if isinstance(l, dict) and str(l.get("fehlt", "")).strip()]
ballast = [str(b).strip() for b in data.get("ballast", []) if str(b).strip()]
return {"ziele": ziele, "luecken": luecken, "ballast": ballast}
def _problems_schema(data):
"""Lese-Check: {"ok":true} → [] · {"problems":[{section,problem}]} → [problem…]."""
if not isinstance(data, dict):
return None
if data.get("ok") is True:
return []
probs = data.get("problems")
if not isinstance(probs, list) or not probs:
return None
out = [str(p.get("problem", "")).strip() for p in probs
if isinstance(p, dict) and str(p.get("problem", "")).strip()]
return out or None
lese = [str(p.get("problem", "")).strip() for p in data.get("lese_probleme", [])
if isinstance(p, dict) and str(p.get("problem", "")).strip()]
return {"claims": claims, "ziele": ziele, "luecken": luecken, "ballast": ballast,
"lese_probleme": lese}
def _first_section(md: str) -> dict | None:
@@ -335,11 +333,6 @@ async def _stage_writer(env: _Env, card: dict) -> bool:
norm = card["block_norm"]
ziele = await db.list_lernziele(env.topic, norm)
ziele_text = "\n".join(f"- ({z['ziel_id']}) {z['text']}" for z in ziele) or "(keine definiert)"
gaps = ""
if card["writer_rounds"] > 0 and card.get("gate_info"):
gaps = ("\nREVISION ROUND — a previous version exists. Revise it: close exactly the gaps "
"below, cut the listed ballast, keep everything else as-is.\n"
f"PREVIOUS VERSION:\n{card.get('md', '')}\n\nGAPS/BALLAST:\n{card['gate_info']}\n")
# oversized first drafts: two halves, merged into one canonical section
if card["writer_rounds"] == 0 and len(env.subs_by_title.get(card["block"], [])) > WRITER_SPLIT_SUBS:
text = await _write_split(env, card, ziele_text)
@@ -348,7 +341,7 @@ async def _stage_writer(env: _Env, card: dict) -> bool:
if text is None:
await _set(env, card, status="error", gate_info="Writer (Split) ohne Ergebnis")
return False
await _set(env, card, md=text, stage="fakten_gate", status="open")
await _set(env, card, md=text, stage="pruefer", status="open")
return True
path = env.slot(f"card-{_safe(norm)}-r{card['writer_rounds']}.md")
@@ -366,7 +359,7 @@ async def _stage_writer(env: _Env, card: dict) -> bool:
assignment=_card_assignment(env, card), ziele=ziele_text,
facts=_card_facts(env, card["block"]),
examples=await _card_examples(env, norm, env.subs_by_title.get(card["block"], [])),
gaps=gaps, spec=env.spec,
gaps="", spec=env.spec,
budget=_writer_budget(len(env.subs_by_title.get(card["block"], []))),
out_path=path, extra=_extra(env.instructions)),
role="guide", capabilities="files", payload=_payload,
@@ -376,190 +369,169 @@ async def _stage_writer(env: _Env, card: dict) -> bool:
if status == FAILED:
await _set(env, card, status="error", gate_info="Writer ohne Ergebnis")
return False
await _set(env, card, md=text, stage="fakten_gate", status="open")
await _set(env, card, md=text, stage="pruefer", status="open")
return True
async def _stage_fakten_gate(env: _Env, card: dict) -> bool:
norm = card["block_norm"]
sec = _first_section(card["md"])
if sec is None:
await _set(env, card, status="error", gate_info="Writer-Fragment unlesbar")
return False
facts = _card_facts(env, card["block"])
ex = await _card_examples(env, norm, env.subs_by_title.get(card["block"], []))
if ex: # the fix agent sees the same facts variable — examples survive the fix pass
facts += "\n\nVERIFIED WORKED EXAMPLES (count as verified facts for this check):\n" + ex
path = env.slot(f"gate-{_safe(norm)}-r{card['writer_rounds']}.json")
status, claims = await run_single_slot(
env.ctx, f"Fakten-Gate {card['block']}", key=f"{env.guide_id}-gate-{_safe(norm)}-r{card['writer_rounds']}",
prompt=_prompt("Guide-Fakten-Gate", topic=env.topic, block=card["block"],
section=sec["md"], facts=facts, out_path=path,
extra=_extra(env.instructions)),
role="judge", capabilities="files",
payload=lambda result: _gate_schema(_json_file(path)),
timeout=_timeout("fakten_gate", 1))
if status == CANCELLED:
return False
if status == FAILED:
claims = [] # gate failure must not block the card — logged, text stands
_log(env.topic, f"Fakten-Gate {card['block']}: kein Ergebnis — Text bleibt ungeprüft")
falsch = [c for c in claims if c.get("urteil") == "falsch"] if claims else []
if claims and not falsch and len(claims) < GATE_FIX_MIN:
# 1–2 merely UNSUPPORTED claims don't justify a fix pass (it ran for 19/20 blocks,
# 40 agent-minutes) — but a WRONG claim always does: one slipped through this
# threshold and cost the guide 1.5 QA points
_log(env.topic, f"Fakten-Gate {card['block']}: {len(claims)} Claim(s) unter Schwelle — kein Fix")
claims = []
if claims:
_log(env.topic, f"Fakten-Gate {card['block']}: {len(claims)} Claims ({len(falsch)} falsch) → Fix")
fixp = env.slot(f"gatefix-{_safe(norm)}-r{card['writer_rounds']}.md")
fixp.unlink(missing_ok=True)
claims_text = "\n".join(f"- {c['text']}" + (f" ({c['grund']})" if c["grund"] else "")
for c in claims)
def _fixload(result):
text = fixp.read_text(encoding="utf-8") if fixp.exists() else ""
return text if _first_section(text) else None
fstatus, fixed = await run_single_slot(
env.ctx, f"Fakten-Fix {card['block']}", key=f"{env.guide_id}-gatefix-{_safe(norm)}-r{card['writer_rounds']}",
prompt=_prompt("Guide-Fakten-Fix", topic=env.topic, block=card["block"],
section=card["md"], claims=claims_text, facts=facts,
out_path=fixp, extra=_extra(env.instructions)),
role="guide", capabilities="files", payload=_fixload, # Fix auf der Schreib-Rolle, Gate auf der Judge-Rolle
timeout=_timeout("fakten_gate", 1))
if fstatus == CANCELLED:
return False
if fstatus == OK and fixed:
new_sec = _first_section(fixed)
# marker invariant: a fix that loses the sub markers kills the level filter → discard
if sec.get("subs") and not (new_sec and new_sec.get("subs")):
_log(env.topic, f"Fakten-Fix {card['block']} ohne Sub-Marker — verworfen")
else:
card["md"] = fixed
await _set(env, card, md=card["md"], stage="coverage", status="open")
return True
async def _stage_coverage(env: _Env, card: dict) -> bool:
norm = card["block_norm"]
ziele = await db.list_lernziele(env.topic, norm)
if not ziele:
await _set(env, card, stage="lesbarkeit")
return True
sec = _first_section(card["md"])
ziele_text = "\n".join(f"- ({z['ziel_id']}) {z['text']}" for z in ziele)
path = env.slot(f"coverage-{_safe(norm)}-r{card['writer_rounds']}.json")
ids = {z["ziel_id"] for z in ziele}
status, res = await run_single_slot(
env.ctx, f"Coverage {card['block']}", key=f"{env.guide_id}-cov-{_safe(norm)}-r{card['writer_rounds']}",
prompt=_prompt("Guide-Coverage", topic=env.topic, block=card["block"],
ziele=ziele_text, section=sec["md"] if sec else card["md"],
out_path=path, extra=_extra(env.instructions)),
role="judge", capabilities="files",
payload=lambda result: _coverage_schema(_json_file(path), ids),
timeout=_timeout("coverage", len(ziele)))
if status == CANCELLED:
return False
if status == FAILED:
_log(env.topic, f"Coverage {card['block']}: kein Ergebnis — weiter ohne Gate")
await _set(env, card, stage="lesbarkeit")
return True
for zid, ok in res["ziele"].items():
if zid in ids:
await db.set_ziel_covered(env.topic, norm, zid, ok)
if res["luecken"] and card["writer_rounds"] < MAX_WRITER_ROUNDS:
info = "\n".join(f"- Lücke ({l['ziel']}): {l['fehlt']}" for l in res["luecken"])
if res["ballast"]:
info += "\n" + "\n".join(f"- Ballast (kürzen): {b}" for b in res["ballast"])
_log(env.topic, f"Coverage {card['block']}: {len(res['luecken'])} Lücke(n) → Writer-Runde "
f"{card['writer_rounds'] + 1}")
await _set(env, card, writer_rounds=card["writer_rounds"] + 1, gate_info=info,
stage="writer", status="open")
return True
if res["luecken"]:
_log(env.topic, f"Coverage {card['block']}: Lücken bleiben nach {MAX_WRITER_ROUNDS} Runden")
await _set(env, card, gate_info="", stage="lesbarkeit")
return True
async def _stage_lesbarkeit(env: _Env, card: dict) -> bool:
norm = card["block_norm"]
sec = _first_section(card["md"])
if sec is None:
await _set(env, card, status="error", gate_info="Fragment unlesbar")
return False
problems: list[str] = []
path = env.slot(f"lese-{_safe(norm)}-r{card['writer_rounds']}.json")
# Text-Antwort + Engine-Sink: Datei-schreibende Judges lieferten invalides JSON
# (3 kaputte Check-Dateien im Messlauf) — der Sink validiert vor dem Persistieren
status, res = await run_single_slot(
env.ctx, f"Lese-Check {card['block']}", key=f"{env.guide_id}-lese-{_safe(norm)}",
prompt=_prompt("Guide-Lese-Check", topic=env.topic, format_name=env.format,
spec=env.spec, sections=f"SECTION: {card['block']}\n{sec['md']}",
extra=_extra(env.instructions)),
role="judge", capabilities="none",
payload=lambda result: _sink_json(result, path, _problems_schema),
timeout=_timeout("lese_check", 1))
if status == CANCELLED:
return False
if status == OK and res:
problems += res
if READABILITY_ACTIVE: # deterministic gate, external grounding
hints = await asyncio.to_thread(readability.rate_sections, {1: sec["md"]})
if hints.get(1):
problems.append(hints[1])
# deterministic length trigger, same formula as the QA detector: prompt guidelines
# alone left writers 2.7–4.1× over target — a measured overshoot forces the fix pass
def _det_hinweise(env: _Env, card: dict, sec: dict) -> list[str]:
"""Deterministische Befunde (extern geerdet, kein LLM): Readability-Modell +
Längen-Rahmen — dieselbe Formel wie der QA-Detektor. Gehen direkt in den Fix
und als „nicht wiederholen"-Notiz in den Prüfer-Prompt."""
out: list[str] = []
subs_all = env.subs_by_title.get(card["block"], [])
n_rel = max(sum(1 for s in subs_all if s.get("relevance") != "peripheral"), 1)
aus = re.split(r"<!--\s*ausführlich\s*-->", sec["md"], maxsplit=1)
pro_sub = len(aus[1] if len(aus) == 2 else sec["md"]) / n_rel
if not (GUIDE_LAENGE_MIN <= pro_sub <= GUIDE_LAENGE_MAX * 0.9):
ziel = _writer_budget(len(subs_all))
problems.append(
out.append(
f"Länge {round(pro_sub)} Zeichen/Sub (Rahmen {GUIDE_LAENGE_MIN}–{round(GUIDE_LAENGE_MAX * 0.9)}): "
f"schreibe den ausführlich-Teil auf etwa {ziel} Zeichen GESAMT um — Sockel-Prosa und "
f"Wiederholungen streichen, alle Sub-Marker und Beispiele behalten")
if problems:
from guide import _level_label
subs = env.subs_by_title.get(card["block"], [])
sub_list = "\n".join(f"- [{_level_label(s)}] {s['title']}" for s in subs) or "(none)"
tasks = (f"SECTION: {card['block']}\n"
f"SUBBLOCKS (set one `<!-- sub: LABEL | title -->` marker each, label/order as here):\n{sub_list}\n"
f"LENGTH TARGET: about {_writer_budget(len(subs))} characters for the detailed "
f"version (guideline — covering every subblock beats brevity).\n"
f"PROBLEM: {' · '.join(problems)}\nCURRENT CONTENT:\n{sec['md']}")
fixp = env.slot(f"lesefix-{_safe(norm)}.md")
fixp.unlink(missing_ok=True)
return out
def _fixload(result):
text = fixp.read_text(encoding="utf-8") if fixp.exists() else ""
return text if _first_section(text) else None
fstatus, fixed = await run_single_slot(
env.ctx, f"Lese-Fix {card['block']}", key=f"{env.guide_id}-lesefix-{_safe(norm)}",
prompt=_prompt("Guide-Sections-Fix", topic=env.topic, format_name=env.format,
facts=_card_facts(env, card["block"]), spec=env.spec, tasks=tasks,
out_path=fixp, extra=_extra(env.instructions)),
role="guide", capabilities="files", payload=_fixload,
timeout=_timeout("writer", 1))
if fstatus == CANCELLED:
async def _det_readability(sec: dict) -> list[str]:
if not READABILITY_ACTIVE:
return []
hints = await asyncio.to_thread(readability.rate_sections, {1: sec["md"]})
return [hints[1]] if hints.get(1) else []
def _auftraege(verdict: dict, det: list[str]) -> tuple[list[str], bool]:
"""Prüfer-Verdikt → Fix-Auftragszeilen. kritisch = falsch-Claims oder Lücken
(nur die rechtfertigen den Re-Prüfer-Pass — Fakten/Coverage sind der Qualitätskern).
Claims-Schwelle wie beim alten Gate: 1–2 nur-„unbelegt" lohnen keinen Fix-Pass."""
claims = verdict["claims"]
falsch = [c for c in claims if c["urteil"] == "falsch"]
if claims and not falsch and len(claims) < GATE_FIX_MIN:
claims = []
zeilen = [f"- CLAIM ({c['urteil']}): {c['text']}" + (f" — {c['grund']}" if c['grund'] else "")
for c in claims]
zeilen += [f"- LÜCKE ({l['ziel']}): {l['fehlt']}" for l in verdict["luecken"]]
zeilen += [f"- BALLAST (kürzen): {b}" for b in verdict["ballast"]]
zeilen += [f"- LESBARKEIT: {p}" for p in verdict["lese_probleme"]]
zeilen += [f"- LESBARKEIT: {p}" for p in det]
return zeilen, bool(falsch or verdict["luecken"])
async def _pruefer_call(env: _Env, card: dict, sec: dict, tag: str, det: list[str]) -> dict | None:
"""EIN Judge-Call prüft Fakten + Coverage + Lesbarkeit (die drei lasen vorher denselben
Section-Text in drei seriellen Calls). Text-Antwort + Engine-Sink (Datei-schreibende
Judges lieferten invalides JSON). → Verdikt | None (FAILED/CANCELLED)."""
norm = card["block_norm"]
ziele = await db.list_lernziele(env.topic, norm)
ziele_text = "\n".join(f"- ({z['ziel_id']}) {z['text']}" for z in ziele) or "(keine definiert)"
ids = {z["ziel_id"] for z in ziele}
facts = _card_facts(env, card["block"])
ex = await _card_examples(env, norm, env.subs_by_title.get(card["block"], []))
if ex: # der Fix sieht dieselben Facts — Beispiele überleben den Fix-Pass
facts += "\n\nVERIFIED WORKED EXAMPLES (count as verified facts for this check):\n" + ex
hinweise = ("\nALREADY NOTED deterministically (do NOT repeat, they go to the fix anyway):\n"
+ "\n".join(f"- {d}" for d in det) + "\n") if det else "\n"
path = env.slot(f"pruefer-{_safe(norm)}-{tag}.json")
status, verdict = await run_single_slot(
env.ctx, f"Prüfer {card['block']}", key=f"{env.guide_id}-pruef-{_safe(norm)}-{tag}",
prompt=_prompt("Guide-Pruefer", topic=env.topic, block=card["block"],
section=sec["md"], facts=facts, ziele=ziele_text, spec=env.spec,
hinweise=hinweise, extra=_extra(env.instructions)),
role="judge", capabilities="none",
payload=lambda result: _sink_json(result, path, lambda d: _pruefer_schema(d, ids)),
timeout=_timeout("fakten_gate", 1))
if status != OK or verdict is None:
return None
for zid, ok in verdict["ziele"].items():
if zid in ids:
await db.set_ziel_covered(env.topic, norm, zid, ok)
return verdict
async def _stage_pruefer(env: _Env, card: dict) -> bool:
"""Verschmolzener Qualitäts-Pass: Fakten-Gate + Coverage + Lese-Check in EINEM Call
(vorher 3 serielle Judges + bis zu 3 Edit-Pässe, die einander überschrieben und deren
letzter ungeprüft blieb). Befunde → Fix-Stage; ohne Befund → done."""
sec = _first_section(card["md"])
if sec is None:
await _set(env, card, status="error", gate_info="Writer-Fragment unlesbar")
return False
det = (await _det_readability(sec)) + _det_hinweise(env, card, sec)
verdict = await _pruefer_call(env, card, sec, f"r{card['writer_rounds']}", det)
if verdict is None:
if is_guide_cancelled(env.guide_id):
return False
if fstatus == OK and fixed:
new_sec = _first_section(fixed)
if sec.get("subs") and not (new_sec and new_sec.get("subs")):
_log(env.topic, f"Lese-Fix {card['block']} ohne Sub-Marker — verworfen")
else:
card["md"] = fixed
await _set(env, card, md=card["md"], stage="done", status="ok", gate_info="")
# fail-open wie das alte Gate: Karte nie blockieren — deterministische Befunde
# gehen trotzdem in den Fix
_log(env.topic, f"Prüfer {card['block']}: kein Ergebnis — nur deterministische Checks")
verdict = {"claims": [], "ziele": {}, "luecken": [], "ballast": [], "lese_probleme": []}
zeilen, kritisch = _auftraege(verdict, det)
if not zeilen:
await _set(env, card, md=card["md"], stage="done", status="ok", gate_info="")
return True
_log(env.topic, f"Prüfer {card['block']}: {len(zeilen)} Befund(e){' (kritisch)' if kritisch else ''} → Fix")
await _set(env, card, gate_info=("KRITISCH\n" if kritisch else "") + "\n".join(zeilen),
stage="fix", status="open")
return True
async def _stage_fix(env: _Env, card: dict) -> bool:
"""EIN kompletter Section-Rewrite unter allen Auflagen (ersetzt Fakten-Fix +
Writer-Revision + Lese-Fix). Danach GENAU EIN Re-Prüfer-Pass, wenn der Fix wegen
falsch-Claims/Lücken lief — der alte Lese-Fix blieb ungeprüft. Rest-Befunde bleiben
sichtbar (gate_info), keine weitere Fix-Runde."""
from guide import _level_label
norm = card["block_norm"]
sec = _first_section(card["md"])
if sec is None:
await _set(env, card, status="error", gate_info="Fragment unlesbar")
return False
info = card.get("gate_info") or ""
kritisch = info.startswith("KRITISCH\n")
auftraege = info.removeprefix("KRITISCH\n")
subs = env.subs_by_title.get(card["block"], [])
sub_list = "\n".join(f"- [{_level_label(s)}] {s['title']}" for s in subs) or "(none)"
fixp = env.slot(f"fix-{_safe(norm)}-r{card['writer_rounds']}.md")
fixp.unlink(missing_ok=True)
def _fixload(result):
text = fixp.read_text(encoding="utf-8") if fixp.exists() else ""
return text if _first_section(text) else None
fstatus, fixed = await run_single_slot(
env.ctx, f"Fix {card['block']}", key=f"{env.guide_id}-gfix-{_safe(norm)}-r{card['writer_rounds']}",
prompt=_prompt("Guide-Fix", topic=env.topic, format_name=env.format, block=card["block"],
section=card["md"], facts=_card_facts(env, card["block"]), spec=env.spec,
auftraege=auftraege, sub_list=sub_list, out_path=fixp,
extra=_extra(env.instructions)),
role="guide", capabilities="files", payload=_fixload,
timeout=_timeout("writer", 1))
if fstatus == CANCELLED:
return False
angewandt = False
if fstatus == OK and fixed:
new_sec = _first_section(fixed)
# marker invariant: a fix that loses the sub markers kills the level filter → discard
if sec.get("subs") and not (new_sec and new_sec.get("subs")):
_log(env.topic, f"Fix {card['block']} ohne Sub-Marker — verworfen")
else:
card["md"] = fixed
angewandt = True
rest = ""
if kritisch and angewandt:
sec2 = _first_section(card["md"])
verdict = await _pruefer_call(env, card, sec2, "re", [])
if verdict is None and is_guide_cancelled(env.guide_id):
return False
if verdict:
zeilen, _k = _auftraege(verdict, [])
if zeilen:
rest = "Rest-Befunde nach Fix:\n" + "\n".join(zeilen)
_log(env.topic, f"Re-Prüfer {card['block']}: {len(zeilen)} Rest-Befund(e) bleiben")
await _set(env, card, md=card["md"], stage="done", status="ok", gate_info=rest)
return True
_STAGE_FN = {"lernziele": _stage_lernziele, "zuweisung": _stage_zuweisung,
"writer": _stage_writer, "fakten_gate": _stage_fakten_gate,
"coverage": _stage_coverage, "lesbarkeit": _stage_lesbarkeit}
"writer": _stage_writer, "pruefer": _stage_pruefer, "fix": _stage_fix}
async def _run_card(env: _Env, card: dict, sem: asyncio.Semaphore) -> None: