"""Board 2 „Artefakte": per finished block, prepare the learning artefacts the guide presents. A card is spawned by board 1's `done` column per mirrored block and runs through: generate → verify (inkl. Fix-Tail) → artefakte (Gen + Prüfer) → finalize (die verschmolzenen Calls liegen in block_calls.py — 4–5 serielle Segmente statt ~20). finalize (SERIAL) merges the block's results into the global sidecar/facts/pattern/artefakte files + the DB tables. Danach eine topic-weite BARRIERE: `outline` (prerequisite graph → chapter order), re-run once per generation run.""" import asyncio import json import logging import re import database as db import blocks from block_calls import _artefakte_block, _generate_block, _verify_block from blocks import ARTEFACT_TYPES, _match_sub, _outline_block from fsutil import atomic_write_json from jsonio import read_json_file as _json_file from kanban import Flow, Stage from pipeline import GenContext, _log from textkit import _norm_title, _title log = logging.getLogger("creator.board_artefacts") BOARD = "artefacts" DONE = "done_artefact" def _nset(msg: str, step: int | None = None) -> None: """Progress no-op — the kanban board itself is the progress display.""" def _safe(norm: str) -> str: return re.sub(r"\W+", "-", norm).strip("-")[:24] or "block" def _sub_key(existing: set[str], sn: str) -> str: """Agents echo the short sub title while the sub row is keyed 'kurztitel: beschreibung' — resolve to the stored key: exact, unambiguous prefix, then unambiguous substring containment either way (agents paraphrase/truncate, measured 23 orphans of ~560 rows). Ambiguous or unresolvable echoes stay unchanged (visible as QA orphan).""" if sn in existing: return sn hits = [s for s in existing if s.startswith(sn + ":")] if len(hits) == 1: return hits[0] if not hits: hits = [s for s in sorted(existing) if sn in s or s in sn] if len(hits) == 1: return hits[0] return sn def _card_set_p(flow: Flow, norm: str): """Per-card progress: the inner step messages land in-memory on the flow — board_snapshot shows them as the card's info line + phase stepper while active. The step INDEX is resolved to its NAME at write time (indices shift with the source type, names are stable).""" info = flow.state.setdefault("card_info", {}) def set_p(msg: str, step: int | None = None) -> None: name = "" if step is not None: steps = flow.state.get("blocks_steps") if steps is None: steps = flow.state["blocks_steps"] = blocks._blocks_steps(flow.topic) if 0 <= step < len(steps): name = steps[step] info[f"{BOARD}:{norm}"] = {"msg": msg, "step": name} return set_p def _pfiles(files: dict, norm: str) -> dict: """Per-block file namespace: own work dir + facts/artefakte paths, global rest.""" sub = files["arbeit"] / f"ab-{_norm_title(norm).replace(' ', '_')[:60]}" sub.mkdir(parents=True, exist_ok=True) return {**files, "arbeit": sub, "facts": sub / "facts.json", "artefakte": sub / "artefakte.json"} def _entry_line(p: dict) -> str: d = p.get("description") return f"{p['title']} — {d}" if d else p["title"] def make_spawner(topic: str, files: dict): """Hook for board 1's `done` column: one artefact card per mirrored block.""" async def spawn(block_card_id: str, payload: dict): norm = payload.get("mirrored_norm") if not norm: return await db.kanban_upsert_card(topic, BOARD, norm, "ablock", "generate", { "title": payload.get("title", ""), "description": payload.get("description", ""), "n_size": payload.get("n_size", 0), # LPT estimate until subs_n exists }) return spawn async def seed_artefact_cards(topic: str, files: dict) -> int: """Board-2-only (Inventar läuft NICHT mit): für jeden fertigen Block eine ablock-Karte erzeugen, falls noch keine existiert — sonst liefe ohne den Board-1-Spawn-Hook nichts durch Board 2. Vorhandene Karten bleiben (sie resümieren in ihrer Spalte). → neue Karten.""" spawn = make_spawner(topic, files) existing = {c["card_id"] for c in await db.kanban_cards(topic, board=BOARD, kind="ablock")} n = 0 for c in await db.kanban_cards(topic, board="inventory", kind="block", stage="done_block"): norm = c["payload"].get("mirrored_norm") if norm and norm not in existing: await spawn(c["card_id"], c["payload"]) n += 1 return n async def _gather_cards(ctx: GenContext, flow: Flow, cards, one): results = await asyncio.gather(*[one(c) for c in cards], return_exceptions=True) errs = [r for r in results if isinstance(r, Exception)] if errs: for e in errs[1:]: # nur errs[0] wird re-raised — der Rest darf nicht stumm verschwinden log.error("weitere Karten-Exception im Batch: %r", e) raise errs[0] flow.wake.set() def _fail_or_cancel(ctx: GenContext, what: str): # A per-card failure belongs on the card (last_error/dead-letter), never in the # topic banner — the inner block functions may have set it there. blocks._blocks_errors.pop(ctx.topic, None) if ctx.is_cancelled(): return None # leave the card where it is raise RuntimeError(f"{what} ohne Ergebnis") # ── Stage processors (one call per card, all parallel) ───────────────────────────── async def _seed_map(topic: str) -> dict[str, list[str]]: """Demoted fragments become seed candidates of their SURVIVING parent block. parent_norm may point at a block that itself got grouped/merged/renamed — follow the redirect chain (grouped → merged_into, rejected → parent_norm, done → mirrored_norm) to the living board-2 card id (= mirrored_norm). A dead end drops the seed (as before).""" alive: set[str] = set() redirect: dict[str, str] = {} rejected: list[dict] = [] for r in await db.kanban_cards(topic, board="inventory", kind="block"): p = r["payload"] tn = _norm_title(p.get("title", "")) if not tn: continue if r["stage"] in ("done", "done_block"): mn = p.get("mirrored_norm") or tn alive.add(mn) if tn != mn: redirect.setdefault(tn, mn) elif r["stage"] == "grouped" and p.get("merged_into"): redirect.setdefault(tn, _norm_title(p["merged_into"])) # umbrella members are absorbed WHOLE topics ("Aufgabenlisten" → "Listen") — # without a seed the umbrella's finders may simply miss them (measured). rejected.append({"title": p.get("title", ""), "parent_norm": _norm_title(p["merged_into"])}) elif r["stage"] == "rejected": if p.get("parent_norm"): redirect.setdefault(tn, p["parent_norm"]) rejected.append(p) def _resolve(norm: str) -> str | None: seen: set[str] = set() cur = norm while cur and cur not in seen: if cur in alive: # alive check BEFORE following (self-edges like „Listen"→„Listen") return cur seen.add(cur) cur = redirect.get(cur, "") return None seeds: dict[str, list[str]] = {} for p in rejected: pn = p.get("parent_norm") if pn and (target := _resolve(pn)): seeds.setdefault(target, []).append(p.get("title", "")) return seeds def _melder(flow: Flow, norm: str): """Live-Stepper der Karte: Phasenname → card_info (ging bei der Call-Verschmelzung verloren — Karten liefen ohne Badge/Stepper durch das Board).""" set_p = _card_set_p(flow, norm) def melde(schritt: str) -> None: try: set_p(f"{schritt}…", step=blocks._step_idx(flow.topic, schritt)) except ValueError: set_p(f"{schritt}…") return melde async def _proc_generate(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards): """Verschmolzener Erzeuger: 2 unabhängige Generatoren liefern Subs+Facts+Einstufung in EINEM Call, Konsens im Code (block_calls._generate_block).""" topic = flow.topic # fragments demoted to a parent become seed candidates of the parent's subblocks seeds = await _seed_map(topic) async def one(c): p = c["payload"] norm = c["card_id"] gen = await _generate_block(ctx, _pfiles(files, norm), p.get("title", ""), p.get("description", ""), instructions, ns=f"{_safe(norm)}-", lbl=f"{p.get('title', norm)} · ", sources=p.get("sources"), seeds=[s for s in seeds.get(norm, []) if s] or None, melde=_melder(flow, norm)) if gen is None: return _fail_or_cancel(ctx, f"Generate {p.get('title', norm)}") p["raw"], p["facts"] = gen["raw"], gen["facts"] p["unsicher"], p["votes"] = gen["unsicher"], gen["votes"] p["subs_n"] = sum(len(v) for v in gen["raw"].values()) + len(gen["unsicher"]) await db.kanban_set_payload(topic, BOARD, norm, p) await db.kanban_advance(topic, BOARD, norm, "verify") await _gather_cards(ctx, flow, cards, one) async def _proc_verify(ctx: GenContext, flow: Flow, files: dict, q: dict, instructions: str, cards): """Verschmolzener Prüfer: MECE + Facts + Einstufung in einem Panel-Call, Fix-Tail inklusive (block_calls._verify_block) — ersetzt facts-check/konsolidierung/levels/ relevance als eigene Stages.""" topic = flow.topic async def one(c): p = c["payload"] norm = c["card_id"] title = p.get("title", "") gen = {"raw": p.get("raw") or {}, "facts": p.get("facts") or {}, "unsicher": p.get("unsicher") or [], "votes": p.get("votes") or {}} res = await _verify_block(ctx, _pfiles(files, norm), title, gen, q, instructions, ns=f"{_safe(norm)}-", lbl=f"{title or norm} · ", sources=p.get("sources"), melde=_melder(flow, norm), keep_all=True) if res is None: return None # nur Cancel — Karte bleibt liegen p["raw"], p["facts"], p["sidecar"] = res["raw"], res["facts"], res["sidecar"] p.pop("unsicher", None) p.pop("votes", None) await db.kanban_set_payload(topic, BOARD, norm, p) await db.kanban_advance(topic, BOARD, norm, "artefakte") await _gather_cards(ctx, flow, cards, one) async def _proc_artefakte(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards): """Fragen + Flashcards + Beispiele in einem Generator-Call, ein Prüfer-Call dahinter (block_calls._artefakte_block).""" topic = flow.topic async def one(c): p = c["payload"] norm = c["card_id"] title = p.get("title", "") res = await _artefakte_block(ctx, _pfiles(files, norm), title, (p.get("sidecar") or {}).get(title) or [], instructions, ns=f"{_safe(norm)}-", lbl=f"{title or norm} · ", melde=_melder(flow, norm)) if res is None: return None # nur Cancel p["pattern"] = res["pattern"] p["artefacts"] = res["artefacts"] await db.kanban_set_payload(topic, BOARD, norm, p) await db.kanban_advance(topic, BOARD, norm, "finalize") await _gather_cards(ctx, flow, cards, one) # ── Finalize (SERIAL): merge into the global files + DB tables ───────────────────── def _merge_json(path, block_keys: dict) -> None: data = _json_file(path) if not isinstance(data, dict): data = {} data.update(block_keys) atomic_write_json(path, data, indent=1) async def _proc_finalize(ctx: GenContext, flow: Flow, files: dict, cards): topic = flow.topic for c in cards: p = c["payload"] title = p.get("title", "") sidecar = p.get("sidecar") or {} pattern = p.get("pattern") or {} artefacts = p.get("artefacts") or {} # global sidecar files (the legacy read path of guide/frontend/resume) _merge_json(files["sub_roh"], {t: subs for t, subs in (p.get("raw") or {}).items()}) _merge_json(files["facts"], p.get("facts") or {}) _merge_json(files["sidecar"], sidecar) _merge_json(files["question_pattern"], pattern) art_global = _json_file(files["artefakte"]) if not isinstance(art_global, dict): art_global = {} for typ in ARTEFACT_TYPES: kept = [e for e in art_global.get(typ, []) if _norm_title(_title(str(e.get("block", "")))) != _norm_title(title)] art_global[typ] = kept + list(artefacts.get(typ, [])) atomic_write_json(files["artefakte"], art_global, indent=1) # DB mirrors — per block only (no global deletes) await blocks._mirror_sidecar_db(topic, sidecar) # stale question/artefact rows of a PREVIOUS run keyed to gone subs: finalize only # upserts, so re-runs left orphans (measured: 28). await db.delete_question_pattern(topic, _norm_title(title)) await db.delete_sub_artefakte(topic, _norm_title(title)) # consensus rows of a PREVIOUS run that this run's sidecar no longer carries would # linger without facts/questions/artefacts (measured: 25) — drop them per block; # variant/discarded stay for QA. Then default-level the mirror's own stragglers. for btitle, subs in sidecar.items(): keep = {_norm_title(str(s.get("title", ""))) for s in subs if isinstance(s, dict)} await db.delete_stale_consensus(topic, _norm_title(btitle), keep - {""}) await db.default_subblock_levels(topic, _norm_title(title)) sub_keys: dict[str, set[str]] = {} async def _keys(bnorm: str) -> set[str]: if bnorm not in sub_keys: sub_keys[bnorm] = {r["sub_norm"] for r in await db.list_subblocks(topic, bnorm)} return sub_keys[bnorm] for btitle, entries in pattern.items(): bnorm = _norm_title(btitle) for e in entries if isinstance(entries, list) else []: sub = str(e.get("subblock", "")).strip() sn = _norm_title(sub) question = str(e.get("question", "")).strip() if bnorm and sn and question: sn = _sub_key(await _keys(bnorm), sn) await db.upsert_question_pattern(topic, bnorm, sn, btitle, sub, question) btitles = list(sidecar.keys()) for typ in ARTEFACT_TYPES: for e in artefacts.get(typ, []): bt = _match_sub(str(e.get("block", "")), btitles) bnorm, sn = _norm_title(bt), _norm_title(str(e.get("subblock", ""))) if not bnorm or not sn: continue sn = _sub_key(await _keys(bnorm), sn) data = json.dumps({k: v for k, v in e.items() if k not in ("block", "subblock")}, ensure_ascii=False) await db.put_sub_artifact(topic, bnorm, sn, typ, data, bt, str(e.get("subblock", ""))) await db.kanban_advance(topic, BOARD, c["card_id"], DONE) _log(topic, f"Artefakte fertig: {title}") flow.wake.set() # ── Outline (topic-wide barrier singleton) ───────────────────────────────────────── OUTLINE_CARD = "outline" async def ensure_outline_card(topic: str) -> None: """(Re-)queue the outline singleton — run once per generation run, after everything.""" await db.kanban_upsert_card(topic, BOARD, OUTLINE_CARD, "outline", "outline", {"title": "Gliederung"}) async def _proc_outline(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards): topic = flow.topic done = await db.kanban_cards(topic, board="inventory", stage="done_block") done.sort(key=lambda c: c["updated_at"]) entries = {i: _entry_line(c["payload"]) for i, c in enumerate(done, 1) if c["payload"].get("title")} if entries: # The outline may run BEFORE finalize has merged the global facts.json — feed the # prereq hints of _learning_order from the card payloads instead (complete as soon # as every block passed the facts stage, which the trimmed barrier guarantees). facts_map: dict = {} for bc in await db.kanban_cards(topic, board=BOARD, kind="ablock"): facts_map.update(bc["payload"].get("facts") or {}) fp = flow.work_dir / "outline-facts.json" atomic_write_json(fp, facts_map, indent=1) plan = await _outline_block(ctx, _nset, {**files, "facts": fp}, entries, instructions) if ctx.is_cancelled(): return if isinstance(plan, dict) and plan.get("chapters"): chapters = [ {"title": ch.get("title", "Kapitel"), "blocks": [_title(entries[n]) for n in ch.get("numbers", []) if n in entries]} for ch in plan["chapters"] ] await db.set_outline(topic, json.dumps({"chapters": chapters}, ensure_ascii=False)) await db.kanban_advance_many(topic, BOARD, [(c["card_id"], DONE) for c in cards]) flow.wake.set() # ── Stage list (appended after board 1 in chain order) ───────────────────────────── _ALT_STAGES = ("subblocks", "facts", "levels", "relevance", "question_pattern", "artefacts", "konsolidierung") # abgeschaltete Stage: verwaiste Karten auf generate zurück async def migriere_alt_karten(topic: str) -> int: """Harter Schnitt: Karten der alten Stage-Treppe beim Flow-Start auf `generate` zurücksetzen (Payload auf die Spawn-Felder reduziert — Zwischenstände der alten Struktur sind für die verschmolzenen Calls wertlos). → Anzahl migrierter Karten.""" moves = [] for c in await db.kanban_cards(topic, board=BOARD, kind="ablock"): if c["stage"] in _ALT_STAGES: p = c["payload"] await db.kanban_set_payload(topic, BOARD, c["card_id"], { "title": p.get("title", ""), "description": p.get("description", ""), "n_size": p.get("n_size", 0), "sources": p.get("sources")}) moves.append((c["card_id"], "generate")) if moves: await db.kanban_advance_many(topic, BOARD, moves) return len(moves) def artefact_stages(ctx: GenContext, flow: Flow, files: dict, q: dict, folder, instructions: str) -> list[Stage]: research_done = lambda: flow.research_done # noqa: E731 return [ Stage(BOARD, "generate", lambda cs: _proc_generate(ctx, flow, files, instructions, cs)), Stage(BOARD, "verify", lambda cs: _proc_verify(ctx, flow, files, q, instructions, cs)), Stage(BOARD, "artefakte", lambda cs: _proc_artefakte(ctx, flow, files, instructions, cs)), Stage(BOARD, "finalize", lambda cs: _proc_finalize(ctx, flow, files, cs), serial=True), # Cross-Block-Dedup (konsolidierung) entfällt im Bottom-up: Board 1 hat die Atome # global geclustert/dedupliziert, es gibt keine block-übergreifenden Sub-Dubletten mehr. Stage(BOARD, "outline", lambda cs: _proc_outline(ctx, flow, files, instructions, cs), barrier=True, drain=True, gate=research_done), ]