"""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 zwei topic-weite BARRIEREN: `konsolidierung` (cross-block
sub dedup, faltet per repair.falte_sub) und `outline` (prerequisite graph → chapter order),
re-run once per generation run — outline läuft parallel zur Dedup-Barriere."""

import asyncio
import hashlib
import json
import logging
import re

import database as db
import blocks
import embedding
from block_calls import _artefakte_block, _generate_block, _verify_block
from blocks import ARTEFACT_TYPES, _match_sub, _neg_set, _sink_json, _outline_block
from config import CROSS_CHUNK_PAARE, EMBEDDING_AKTIV, SUB_DUP_KANDIDAT_COS
from fsutil import atomic_write_json
from jsonio import read_json_file as _json_file
from kanban import Flow, Stage
from pipeline import FAILED, GenContext, _extra, _log, _prompt, _timeout, run_single_slot
from textkit import _norm_title, _title, parse_facts

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 _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)


def _cross_schema(data) -> dict[int, str] | None:
    """{"pairs": {"1": "a"|"b"|"nein"}} → {pair_nr: verdict} · otherwise None."""
    if not isinstance(data, dict) or not isinstance(data.get("pairs"), dict):
        return None
    out: dict[int, str] = {}
    for k, v in data["pairs"].items():
        try:
            nr = int(k)
        except (ValueError, TypeError):
            continue
        s = str(v).strip().casefold()
        if s in ("a", "b", "nein"):
            out[nr] = s
    return out or None


async def _proc_konsolidierung(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards):
    """BARRIER/drain am RUN-ENDE — cross-block sub dedup: the SAME statement carried by two
    blocks (measured on Markdown: tab handling in 3 blocks, HTML blocks, backslash escapes —
    the in-block paths never see these). Embedding candidates (block≠block, cos ≥
    SUB_DUP_KANDIDAT_COS) go to a two-judge panel; UNANIMITY decides which block keeps the
    statement. Sitzt seit dem Umbau NACH finalize: als Mittel-Barriere wartete jede fertige
    Karte auf die langsamste (gemessen: 8:46 min Leerlauf pro Block, kanban-smoke). Der
    Verlierer wird per repair.falte_sub gefaltet (variant + Fragen/Artefakte umhängen) —
    die wenigen Cross-Dubletten kosten so ein paar umsonst generierte Artefakte statt
    Minuten Wandzeit für alle. Fail-open on judge failure/dissent."""
    from repair import falte_sub
    topic = flow.topic
    work_dir = flow.work_dir
    # Resume-Karten aus der alten Stage-Position (Barriere lag vor den Fragen): erst fertig
    # generieren — die Barriere feuert erneut, wenn alle wieder hier sind. Direkt dedupen
    # ginge schief: finalize würde den gefalteten Sub aus dem Karten-Sidecar re-spiegeln.
    nachzuegler = [(c["card_id"], "artefakte" if "sidecar" in c["payload"] else "generate")
                   for c in cards if "pattern" not in c["payload"]]
    if nachzuegler:
        await db.kanban_advance_many(topic, BOARD, nachzuegler)
        flow.wake.set()
        return

    async def _advance_all():
        await db.kanban_advance_many(topic, BOARD, [(c["card_id"], DONE) for c in cards])
        flow.wake.set()

    rows = [r for r in await db.list_subblocks(topic) if r["status"] == "consensus"]
    if len(rows) < 2 or not EMBEDDING_AKTIV or not await asyncio.to_thread(embedding.available):
        await _advance_all()
        return
    sims = await asyncio.to_thread(embedding.embed_sims, [r["sub_title"] for r in rows])
    if sims is None:
        await _advance_all()
        return
    negs = [_neg_set(r["sub_title"]) for r in rows]
    pairs = [(i, j) for i in range(len(rows)) for j in range(i + 1, len(rows))
             if rows[i]["block_norm"] != rows[j]["block_norm"] and negs[i] == negs[j]
             and float(sims[i][j]) >= SUB_DUP_KANDIDAT_COS]
    if not pairs:
        await _advance_all()
        return

    def _kp(r: dict) -> list:
        return parse_facts(r.get("facts")).get("key_points") or []

    def _side(tag: str, r: dict) -> str:
        return f"{tag}: [Block: {r['block']}] {r['sub_title']}" + "".join(f"\n   - {p}" for p in _kp(r))

    # chunked judging: ONE call over all pairs scaled its timeout past 50 min, and a hung
    # call blocked the barrier for the full window (measured on aak: 196 pairs, 2×54 min)
    chunks = [pairs[lo:lo + CROSS_CHUNK_PAARE] for lo in range(0, len(pairs), CROSS_CHUNK_PAARE)]

    async def _urteile_chunk(chunk: list[tuple[int, int]]) -> dict[int, str]:
        """Two judges (+ substitute, + tie-breaker) over one pair chunk → {local_k: verdict};
        empty dict = fail-open (pairs stay)."""
        lines = "\n\n".join(
            f"{k}.\n{_side('A', rows[i])}\n{_side('B', rows[j])}"
            for k, (i, j) in enumerate(chunk, 1))
        h = hashlib.md5(lines.encode()).hexdigest()[:8]
        paths = [work_dir / f"sub-crossblock-{h}-j{j}.json" for j in (1, 2)]

        async def _judge(j, path, plines, n):
            if _cross_schema(_json_file(path)) is not None:
                return  # resume
            status, _v = await run_single_slot(
                ctx, f"Sub-Crossblock j{j}", key=f"blocks-{topic}-sub-crossblock-{h}-j{j}",
                prompt=_prompt("Subblock-Crossblock", topic=topic, pairs=plines, extra=_extra(instructions)),
                role="judge", capabilities="none",
                payload=lambda result, p=path: _sink_json(result, p, _cross_schema),
                timeout=_timeout("subblock_check", n))
            if status == FAILED:
                _log(topic, f"Sub-Crossblock j{j} ohne Ergebnis — fail-open")

        await asyncio.gather(*[_judge(j, p, lines, len(chunk)) for j, p in zip((1, 2), paths)])
        if ctx.is_cancelled():
            return {}
        outs = [o for p in paths if (o := _cross_schema(_json_file(p))) is not None]
        if len(outs) == 1:  # Ersatz-Richter statt fail-open bei EINEM Ausfall
            ersatz = work_dir / f"sub-crossblock-{h}-jE.json"
            await _judge("E", ersatz, lines, len(chunk))
            if ctx.is_cancelled():
                return {}
            outs = [o for p in [*paths, ersatz] if (o := _cross_schema(_json_file(p))) is not None]
        if len(outs) != 2:
            if outs:
                _log(topic, "Sub-Crossblock: nur 1/2 Richter — fail-open")
            return {}
        final = {k: (outs[0].get(k, "nein") if outs[0].get(k, "nein") == outs[1].get(k, "nein")
                     else "uneinig") for k in range(1, len(chunk) + 1)}
        disputed = [k for k, v in final.items() if v == "uneinig"]
        if disputed:  # tie-breaker: a third judge sees ONLY the disputed pairs, majority 2/3
            d_lines = "\n\n".join(
                f"{x}.\n{_side('A', rows[chunk[k - 1][0]])}\n{_side('B', rows[chunk[k - 1][1]])}"
                for x, k in enumerate(disputed, 1))
            p3 = work_dir / f"sub-crossblock-{h}-j3.json"
            await _judge(3, p3, d_lines, len(disputed))
            if ctx.is_cancelled():
                return {}
            v3 = _cross_schema(_json_file(p3)) or {}
            if not v3:
                _log(topic, "Sub-Crossblock j3 ohne Ergebnis — strittige Paare bleiben")
            for x, k in enumerate(disputed, 1):
                t = v3.get(x, "nein")
                if t in (outs[0].get(k, "nein"), outs[1].get(k, "nein")):
                    final[k] = t  # majority 2/3; anything else stays disputed → no fold
        return final

    chunk_finals = await asyncio.gather(*[_urteile_chunk(c) for c in chunks])
    if ctx.is_cancelled():
        return
    final_all: dict[int, str] = {}  # global pair index (1-based over `pairs`) → verdict
    for cnr, fin in enumerate(chunk_finals):
        for k, v in fin.items():
            final_all[cnr * CROSS_CHUNK_PAARE + k] = v
    journal = {"paare": len(pairs), "chunks": len(chunks), "gefaltet": [], "verdicts": []}
    gone: set[tuple] = set()
    for k, (i, j) in enumerate(pairs, 1):
        verdict = final_all.get(k, "nein")
        journal["verdicts"].append({"a": f"{rows[i]['block']} · {rows[i]['sub_title']}",
                                    "b": f"{rows[j]['block']} · {rows[j]['sub_title']}",
                                    "verdict": verdict})
        if verdict not in ("a", "b"):
            continue
        win, lose = (rows[i], rows[j]) if verdict == "a" else (rows[j], rows[i])
        wk = (win["block_norm"], win["sub_norm"])
        lk = (lose["block_norm"], lose["sub_norm"])
        if lk in gone or wk in gone:  # keeper already folded → don't chain away the content
            continue
        await falte_sub(topic, files, win, lose)
        gone.add(lk)
        journal["gefaltet"].append({"weg": f"{lose['block']} · {lose['sub_title']}",
                                    "bleibt": f"{win['block']} · {win['sub_title']}"})
    if journal["gefaltet"]:
        _log(topic, f"Sub-Crossblock: {len(journal['gefaltet'])} blockübergreifende Dublette(n) gefaltet")
    hg = hashlib.md5("\n".join(f"{i}:{j}" for i, j in pairs).encode()).hexdigest()[:8]
    atomic_write_json(work_dir / f"sub-crossblock-{hg}.json", journal, indent=1)
    await _advance_all()


# ── 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),
    ]