creator/backend/board_inventory.py

"""Board 1 „Inventar": streaming kanban stages for the block inventory.

Research producers stream candidate titles into the board; the columns pull, filter,
merge and reject them until an optimized block inventory remains. The mature filter
logic (relation guard, complete-link cliques, degrade pass, umbrella grouping, gates)
is reused from blocks.py — this module only re-orchestrates it as streaming stages.

Stages (cards):
  ingest          title    exact dedup folded at DB-add (reader union)
  cluster         title    SERIAL — online embedding nearest-neighbour clustering
  pair_check      cluster  judge per candidate pair (relation guard, cliques) → splits
  consensus_gate  cluster  code: reader union ≥2 (or supplement) → naming, else clarify
  clarify         cluster  3-judge panel, unanimity for single-reader finds
  naming          cluster  judge picks the canonical member title
  naming_check    cluster  second judge verifies → spawns the block card
  fragment_filter block    BARRIER/drain: global re-merge + degrade pass (full list)
  dedup           block    BARRIER/drain: global judge-verified pair dedup (incl. context)
  grouping        block    BARRIER/drain: umbrella grouping (type gate, reconcile)
  gap_check       block    BARRIER/drain: one supplement round (web) → feeds ingest
  done            block    mirror into the legacy `blocks` table → done_block

Terminal: clustered (titles), done_cluster, grouped, rejected (with journal), done_block.
"""

import asyncio
import hashlib
import json
import logging
import math
import re
import shutil
import unicodedata
import uuid
from datetime import datetime, timezone

import database as db
import embedding
import kanban
from kanban import Flow, Stage, chain_stages
import blocks
from blocks import (
    DEDUP_GLOBAL_FLOOR, DEDUP_PAIR_FLOOR, DEDUP_PAIRS_CHUNK, DEDUP_TITLE_AUTO, FILTER_CHUNK,
    FILTER_RECHECK_PANEL, CONSOLIDATION_PANEL, RESEARCH_BATCH, RESEARCH_READERS,
    RESEARCH_THEMA_AGENTS, _FILTER_NOTATION, _GROUP_STANDALONE,
    _build_research_prompt, _canonical, _canonical_key, _chunk_nums, _cliques,
    _completion_schema, _containment_parent, _crawl_index, _file_payload,
    _filter_schema, _filter_suspect, _is_artifact, _is_named_statement,
    _is_parentless_noise, _is_reference, _pairs_schema, _read,
    _relation_conflict, _root, _supplement_schema, _text_sections, _umbrella_schema,
    _aspect_marker, _title_variants, _corpus_files, _evidence_pack, _sink_json, source_folder,
)
from config import (QA_GATE_NOTE, QA_GATE_LLM,
    
    BLOCKS_GRUPPIERUNG_AKTIV, EMBEDDING_AKTIV, EMBEDDING_BLOCK_CAP,
    EMBEDDING_SIBLING_CAP, EMBEDDING_SIBLING_FLOOR, FRAGMENT_MIN_COS,
    GROUP_MIN_COS_FLOOR, GROUP_RECONCILE_FLOOR,
)
from fsutil import atomic_write_json, atomic_write_text
from jsonio import read_json_file as _json_file
from pipeline import (CANCELLED, FAILED, OK, GenContext, _extra, _log, _prompt,
                      _runde_schema, _timeout, _yesno_schema, run_single_slot)
from textkit import _norm_title, _parse_selection, _title, clean_title

log = logging.getLogger("creator.board_inventory")

BOARD = "inventory"
from config import RESEARCH_RUNTIME  # noqa: E402 — zentral tunebar
_POLL_RESEARCH = 3      # seconds between live reads of a running research file

_ingest_lock = asyncio.Lock()  # serializes the read-modify-write title upserts


def _h(*parts: str) -> str:
    """Short stable hash for slot filenames — a rework with different members must not
    reuse a stale judge file (indices would no longer match)."""
    return hashlib.sha1("|".join(parts).encode("utf-8")).hexdigest()[:8]


def _t_text(p: dict) -> str:
    return f"{p['title']} — {p['description']}" if p.get("description") else p["title"]


def _line(i: int, p: dict, mark: str = "") -> str:
    d = p.get("description")
    return f"{i}. {mark}{p['title']} — {d}" if d else f"{i}. {mark}{p['title']}"


def _naming_schema(data, count: int) -> tuple[int | None, str | None, bool] | None:
    """{"ok":true} → (None,None,True) · {"best":N[,"name":…]} → (N, name|None, False).
    name über 80 Zeichen wird verworfen (Kürze ist der Zweck der Abstraktion)."""
    if not isinstance(data, dict):
        return None
    if data.get("ok") is True:
        return None, None, True
    try:
        n = int(data.get("best"))
    except (ValueError, TypeError):
        return None
    if not 1 <= n <= count:
        return None
    name = str(data.get("name", "")).strip() or None
    if name and len(name) > 80:
        name = None
    return n, name, False


def _sanierung_schema(data) -> tuple[str, str] | None:
    """{"title": …, "description": …} → (title, description), beides getrimmt; Titel über
    80 Zeichen verfällt (leerer String = kein Vorschlag, Feld bleibt wie es ist)."""
    if not isinstance(data, dict):
        return None
    t = str(data.get("title") or "").strip()
    d = str(data.get("description") or "").strip()
    if not t and not d:
        return None
    if len(t) > 80:
        t = ""
    return t, d


def _name_verankert(name: str, rows: list[dict], ctoks: set[str] | None) -> bool:
    """Abstraction guard: a free-formed title must be anchored — in the corpus (uni/projekt)
    or in the members' own words (thema). Unanchored names drift to textbook canon
    (measured: 'Königsberger Brückenproblem', 0 corpus hits) → fall back to best-of-members."""
    if ctoks is not None:
        return _hat_anker(name, ctoks)
    from qa import _STOP
    basis = " ".join(f"{r.get('title', '')} {r.get('description') or ''}" for r in rows)
    fold = lambda s: unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode().casefold()  # noqa: E731
    btoks = set(re.findall(r"\w{3,}", fold(basis)))
    ntoks = {t for t in re.findall(r"\w{3,}", fold(name)) if t not in _STOP}
    return bool(ntoks) and ntoks <= btoks


# ── Embedding cache (per flow) ─────────────────────────────────────────────────────
async def _emb_ok(flow: Flow) -> bool:
    if "emb_ok" not in flow.state:
        flow.state["emb_ok"] = EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available)
    return flow.state["emb_ok"]


async def _vec_rows(flow: Flow, texts: list[str]):
    """L2-normalized vectors for `texts`, cached per flow (a title is embedded once,
    not once per stage). → (n, d) ndarray | None when the model is off."""
    if not texts or not await _emb_ok(flow):
        return None
    cache = flow.state.setdefault("vecs", {})
    missing = list(dict.fromkeys(t for t in texts if t not in cache))
    if missing:
        arr = await asyncio.to_thread(embedding.embed, missing)
        if arr is None:
            return None
        for t, v in zip(missing, arr):
            cache[t] = v
    import numpy as np
    return np.vstack([cache[t] for t in texts])


# ── Research producers ─────────────────────────────────────────────────────────────
def _extract_text(raw_line: str) -> str:
    """Best-effort: pull assistant/tool text out of ONE opencode `--format json` event line.
    Recursively collects every `text`/`content` string — robust to the exact event schema."""
    try:
        obj = json.loads(raw_line)
    except Exception:
        return ""
    parts: list[str] = []

    def _walk(o):
        if isinstance(o, dict):
            for k, v in o.items():
                if k in ("text", "content") and isinstance(v, str):
                    parts.append(v)
                else:
                    _walk(v)
        elif isinstance(o, list):
            for v in o:
                _walk(v)
    _walk(obj)
    return "".join(parts)


async def _ingest_titles(flow: Flow, text: str, reader: str, source: str = "") -> int:
    """Parse a reader text into title cards (stage 'ingest'). Exact dupes fold via
    reader union. Repeated drains over a growing buffer are idempotent. → new count."""
    n, seen = 0, set()
    for record in _parse_selection(text).values():
        title = clean_title(_title(record))
        norm = _norm_title(title)
        if not norm or norm in seen:
            continue
        seen.add(norm)  # one reader = one vote per concept
        parts = [t.strip() for t in record.split(" — ")]
        desc = clean_title(parts[1]) if len(parts) >= 2 else ""
        src = source or (parts[2] if len(parts) >= 3 else "")
        async with _ingest_lock:
            if await db.kanban_add_title(flow.topic, BOARD, norm, title, desc, src, reader):
                n += 1
    if n:
        flow.wake.set()
    return n


async def _research_once(ctx: GenContext, flow: Flow, q: dict, folder, instructions: str,
                         tag: str, *, section: str = "", fokus: str = "", source_file: str = ""):
    """ONE agent searches; its titles stream into the ingest queue LIVE. Two sources feed
    the ingest: the JSON event stream (on_line → text buffer) AND the file the agent writes —
    whichever the agent uses, cards stream in immediately (not only after it finishes)."""
    work_dir = flow.work_dir
    caps = "files" if folder else "full"
    p = work_dir / f"research-{tag}.md"
    stop = asyncio.Event()
    buf: list[str] = []  # assistant text streamed live from the JSON events

    def _on_line(raw: str):  # sync, called per stdout line by the agent runner
        if (t := _extract_text(raw)):
            buf.append(t)

    async def _drain() -> bool:  # ingest from BOTH event buffer and file (idempotent)
        text = "".join(buf)
        if (ft := _file_payload(p)):
            text += "\n" + ft
        return bool(text) and await _ingest_titles(flow, text, tag, source_file)

    async def _tail():  # live-ingest loop while the agent runs
        while not stop.is_set():
            try:
                await asyncio.wait_for(stop.wait(), timeout=_POLL_RESEARCH)
            except asyncio.TimeoutError:
                pass
            await _drain()

    if not _file_payload(p):  # resume: a valid reader file is re-ingested without an agent
        p.unlink(missing_ok=True)
        tail = asyncio.create_task(_tail())
        try:
            await run_single_slot(
                ctx, f"Research {tag}", key=f"blocks-{ctx.topic}-research-{tag}",
                prompt=_build_research_prompt(ctx.topic, p, instructions, q["type"], folder,
                                              fokus=fokus, section=section),
                role="quick", capabilities=caps,
                payload=(lambda result, p=p: _file_payload(p)),
                timeout=RESEARCH_RUNTIME, on_line=_on_line,
            )
        finally:
            stop.set()
            await tail
    await _drain()  # final catch-up (also the whole resume path)
    _log(ctx.topic, f"Research {tag}: Titel → ingest")


def _build_producers(ctx: GenContext, flow: Flow, q: dict, folder, instructions: str) -> list:
    """Producer coroutines per source mode. thema = N web agents; uni/projekt = 2 readers per
    text section; link = 2 readers per fixed page batch (coverage ticked per batch)."""
    topic = ctx.topic

    if not folder:  # thema: free web research
        return [_research_once(ctx, flow, q, folder, instructions, str(i))
                for i in range(1, RESEARCH_THEMA_AGENTS + 1)]

    if q["type"] in ("uni", "projekt"):
        sections: list[tuple[str, str]] = []
        for fn in sorted(set(_crawl_index(folder).values()) or
                         [f.name for f in sorted(folder.glob("**/*.txt"))]):
            for sec in _text_sections(_read(folder / fn)):
                sections.append((fn, sec))
        prods = []
        for ei, (fn, sec) in enumerate(sections, 1):
            block = (f"ARBEITE AUSSCHLIESSLICH MIT DIESEM TEXTABSCHNITT. Lies ihn VOLLSTÄNDIG, "
                     f"überspringe nichts. Suche NICHT im Web — nur diese Section zählt."
                     f"\n\n-----\n{sec}\n-----")
            prods += [_research_once(ctx, flow, q, folder, instructions, f"a{ei}-{r}",
                                     section=block, source_file=fn)
                      for r in range(1, RESEARCH_READERS + 1)]
        return prods

    # link/crawl: fixed page batches, RESEARCH_READERS readers each
    async def _batch(bi: int, batch: list[str]):
        liste = "\n".join(f"- {p}" for p in batch)
        fokus = ("WICHTIG — feste Assignment: Bearbeite AUSSCHLIESSLICH diese Dateien und lies JEDE "
                 f"vollständig. Ignoriere alle anderen Dateien im Ordner:\n{liste}")
        try:
            await asyncio.gather(*[
                _research_once(ctx, flow, q, folder, instructions, f"b{bi}-{r}", fokus=fokus)
                for r in range(1, RESEARCH_READERS + 1)], return_exceptions=True)
        finally:
            await db.mark_sources_read_done(topic, batch)  # tick coverage even without hits

    return [_batch(bi, b) for bi, b in enumerate(
        _chunk_nums(sorted(flow.state["pages"]),
                    max(1, math.ceil(len(flow.state["pages"]) / RESEARCH_BATCH))), 1)]


# ── Column processors ──────────────────────────────────────────────────────────────
async def _proc_ingest(flow: Flow, cards):
    """Visibility column — exact dedup already folded at DB-add. Just advance."""
    await db.kanban_advance_many(flow.topic, BOARD, [(c["card_id"], "cluster") for c in cards])
    flow.wake.set()


async def _proc_cluster(ctx: GenContext, flow: Flow, cards):
    """SERIAL online clustering: nearest neighbour over ALL known titles (including the
    batch mates registered a moment ago — fixes main's same-batch blindness). Join the
    neighbour's cluster (cos ≥ floor, cap 25) or open a new one; the touched cluster is
    reworked from pair_check."""
    topic = flow.topic
    emb = await _emb_ok(flow)
    membership = await db.kanban_membership(topic)
    sizes: dict[str, int] = {}
    for g in membership.values():
        sizes[g] = sizes.get(g, 0) + 1
    nn = flow.state.setdefault("nn", {"norms": [], "texts": []})
    moves = []
    for c in cards:
        nm, p = c["card_id"], c["payload"]
        text = _t_text(p)
        target = None
        if emb and nn["norms"]:
            mat = await _vec_rows(flow, nn["texts"] + [text])
            if mat is not None:
                import numpy as np
                cos = mat[:-1] @ mat[-1]
                best = int(np.argmax(cos))
                if float(cos[best]) >= DEDUP_PAIR_FLOOR:
                    cand = membership.get(nn["norms"][best])
                    if cand and sizes.get(cand, 0) < EMBEDDING_BLOCK_CAP:
                        target = cand
        cid = target or f"cl-{uuid.uuid4().hex[:12]}"
        membership[nm] = cid
        sizes[cid] = sizes.get(cid, 0) + 1
        await db.kanban_set_member(topic, nm, cid)
        # gaining a member reworks the cluster from pair_check (live re-clustering)
        await db.kanban_upsert_card(topic, BOARD, cid, "cluster", "pair_check",
                                    {"title": p["title"], "description": p.get("description", "")})
        nn["norms"].append(nm)
        nn["texts"].append(text)
        moves.append((nm, "clustered"))
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _member_rows(topic: str, cid: str) -> list[dict]:
    rows = []
    for nm in await db.kanban_members_of(topic, cid):
        tc = await db.kanban_get_card(topic, BOARD, nm)
        if tc:
            rows.append({"norm": nm, **tc["payload"]})
    return rows


async def _proc_pair_check(ctx: GenContext, flow: Flow, cards):
    results = await asyncio.gather(*[_pair_one(ctx, flow, c) for c in cards], return_exceptions=True)
    errs = [r for r in results if isinstance(r, Exception)]
    if errs:
        raise errs[0]  # unadvanced cards go to backoff (engine)
    flow.wake.set()


async def _pair_one(ctx: GenContext, flow: Flow, c):
    """Pairwise entity resolution inside one cluster: candidate pairs = mean(title, title+desc)
    cosine ≥ floor ∪ canonical-key blocking; a judge confirms each pair; auto-merges (title ≥0.95,
    exact key); relation guard blocks different reductions; complete-link cliques merge — the
    rest splits into fresh clusters. All resulting clusters → consensus_gate."""
    topic = flow.topic
    cid = c["card_id"]
    rows = await _member_rows(topic, cid)
    if len(rows) <= 1:
        await db.kanban_advance(topic, BOARD, cid, "consensus_gate")
        return
    n = len(rows)
    texts_full = [_t_text(r) for r in rows]
    texts_title = [r["title"] for r in rows]
    vf = await _vec_rows(flow, texts_full)
    vt = await _vec_rows(flow, texts_title)
    sims_title = vt @ vt.T if vt is not None else None
    pairs: set[tuple[int, int]] = set()
    if vf is not None and vt is not None:
        sims = (vf @ vf.T + sims_title) / 2
        pairs |= {(i, j) for i in range(n) for j in range(i + 1, n)
                  if float(sims[i][j]) >= DEDUP_PAIR_FLOOR}
    else:  # no model → judge all pairs (clusters are small)
        pairs |= {(i, j) for i in range(n) for j in range(i + 1, n)}
    key_groups: dict[str, list[int]] = {}
    for i, r in enumerate(rows):
        if (k := _canonical_key(r["title"])):
            key_groups.setdefault(k, []).append(i)
    for grp in key_groups.values():
        for x in range(len(grp)):
            for y in range(x + 1, len(grp)):
                pairs.add((grp[x], grp[y]))
    ordered = sorted(pairs)
    edges: list[tuple[int, int]] = []
    chunks = [ordered[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(ordered), DEDUP_PAIRS_CHUNK)]
    h = _h(*[r["norm"] for r in rows])

    async def _judge_pairs(ci, chunk):  # chunks are independent — one parallel wave
        path = flow.work_dir / f"pair-{cid}-{h}-c{ci}.json"
        if _pairs_schema(_json_file(path)) is not None:
            return  # resume
        lines = "\n\n".join(
            f"{j + 1}.\nA: {_t_text(rows[a])}\nB: {_t_text(rows[b])}"
            for j, (a, b) in enumerate(chunk))
        status, _v = await run_single_slot(
            ctx, f"Paar-Check {cid}", key=f"blocks-{topic}-pair-{cid}-{h}-c{ci}",
            prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=path),
            role="judge", capabilities="files",
            payload=lambda result, p=path: _pairs_schema(_json_file(p)),
            timeout=_timeout("selection_mapping", len(chunk)))
        if status == FAILED:
            raise RuntimeError(f"Paar-Filter {cid} chunk {ci} ohne Ergebnis")

    results = await asyncio.gather(*[_judge_pairs(ci, c) for ci, c in enumerate(chunks)],
                                   return_exceptions=True)
    errs = [r for r in results if isinstance(r, Exception)]
    if errs:
        raise errs[0]
    if ctx.is_cancelled():
        return
    for ci, chunk in enumerate(chunks):
        verdict = _pairs_schema(_json_file(flow.work_dir / f"pair-{cid}-{h}-c{ci}.json")) or {}
        for j, (a, b) in enumerate(chunk):
            if verdict.get(j + 1) and not _relation_conflict(rows[a]["title"], rows[b]["title"]):
                edges.append((a, b))
    if sims_title is not None:  # auto recall net: near-identical titles merge without the judge
        for a, b in ordered:
            if float(sims_title[a][b]) >= DEDUP_TITLE_AUTO and not _relation_conflict(
                    rows[a]["title"], rows[b]["title"]):
                edges.append((a, b))
    for grp in key_groups.values():  # exact-canonical-key auto-merge (ER ~100% precision)
        for x in range(len(grp)):
            for y in range(x + 1, len(grp)):
                a, b = grp[x], grp[y]
                if not _relation_conflict(rows[a]["title"], rows[b]["title"]):
                    edges.append((a, b))
    groups = _cliques(n, edges)
    used = set().union(*[set(g) for g in groups]) if groups else set()
    groups += [[i] for i in range(n) if i not in used]
    groups.sort(key=len, reverse=True)
    for gi, g in enumerate(groups):
        gid = cid if gi == 0 else f"cl-{uuid.uuid4().hex[:12]}"
        await db.kanban_set_members(topic, gid, [rows[i]["norm"] for i in g])
        rep = rows[max(g, key=lambda k: len(rows[k].get("description") or ""))]
        await db.kanban_upsert_card(topic, BOARD, gid, "cluster", "consensus_gate",
                                    {"title": rep["title"], "description": rep.get("description", "")})


def _rep(rows: list[dict]) -> dict:
    """Cluster representative via the survivorship cascade (main concept, not an aspect)."""
    cands = [{"title": r["title"], "description": r.get("description") or "",
              "reader": r.get("readers") or []} for r in rows]
    return _canonical(cands, list(range(len(cands))), set())


_ANKER_STOP = {"der", "die", "das", "und", "oder", "für", "mit", "von", "des", "den", "dem",
               "ein", "eine", "einer", "the", "and", "for", "problem", "probleme",
               "algorithmus", "algorithmen", "definition", "satz", "lemma", "beispiel",
               "methode", "verfahren"}


def _korpus_tokens(folder) -> set[str]:
    """All corpus word tokens (≥3 chars, casefolded) — anchor base for the reader-title gate."""
    toks: set[str] = set()
    for f in _corpus_files(folder, None):
        try:
            toks |= set(re.findall(r"\w{3,}", f.read_text(encoding="utf-8").casefold()))
        except OSError:
            continue
    return toks


def _hat_anker(title: str, ctoks: set[str]) -> bool:
    """≥1 distinctive title token appears in the corpus — digit-suffix tolerant:
    '∆TSP1' → 'tsp1' → 'tsp' (the corpus tokenizes '∆TSP' to 'tsp')."""
    for t in re.findall(r"\w{3,}", title.casefold()):
        if t in _ANKER_STOP:
            continue
        forms = {t}
        a = unicodedata.normalize("NFKD", t).encode("ascii", "ignore").decode()
        if len(a) >= 3:
            forms.add(a)  # Symbol-Präfixe (δtsp1 → tsp1) — der Korpus-Tokenizer kennt kein ∆
        forms |= {f2 for f in list(forms) if len(f2 := f.rstrip("0123456789")) >= 3}
        if any(ct == f or ct.startswith(f) for f in forms for ct in ctoks):
            return True
    return False


def _sanierung_noetig(p: dict, ctoks: set[str] | None) -> bool:
    """QA-messbare Befund-Formen am Entstehungsort: Titel ohne Korpus-Anker (QA: fremd —
    misst Token-Anker, nicht Semantik) oder leere Beschreibung (QA: hygiene). Gilt auch für
    Singleton-Cluster, die das Naming sonst überspringen — Reader-Rohtitel gingen wörtlich
    bis done_block durch (gemessen: 'k-Coloring' statt Korpus-Form 'k-Color')."""
    if not (p.get("description") or "").strip():
        return True
    return bool(ctoks) and not _hat_anker(p.get("title", ""), ctoks)


async def _anker_beleg(ctx: GenContext, flow: Flow, kandidaten: list[tuple[str, dict]]) -> set[str]:
    """Evidence judge for quorum titles WITHOUT any corpus anchor — two readers naming the
    same famous canon independently beat the quorum although the material never mentions it
    (measured: 'Königsberger Brückenproblem', 0 corpus hits). FAIL-OPEN for judge errors —
    but an EMPTY evidence pack (no distinctive token anywhere in the corpus) rejects
    deterministically: the judge approved 3 canon titles on sham excerpts. → card_ids to reject."""
    from qa import _distinctive
    topic = flow.topic
    folder = source_folder(topic)
    lines = []
    hart_nein: set[str] = set()
    for k, (cid, p) in enumerate(kandidaten, 1):
        # distinctive tokens only: title+description as raw queries pulled generic sections
        # („Beispiel", „klassisch") — the judge saw sham excerpts and waved canon through
        toks = _distinctive(p.get("title", "")) | _distinctive(p.get("description") or "")
        ev = _evidence_pack(folder, None, [" ".join(sorted(toks))], budget=4000) if toks else ""
        if not ev:  # nothing to attest — deterministic reject, no judge to sweet-talk
            hart_nein.add(cid)
        lines.append(f"{k}. {p.get('title', '')} — {p.get('description') or ''}\nAUSZÜGE:\n"
                     f"{ev or '(keine passenden Auszüge im Material gefunden)'}")
    if len(hart_nein) == len(kandidaten):
        return hart_nein
    h = _h(*[cid for cid, _ in kandidaten])
    path = flow.work_dir / f"anker-beleg-{h}.json"
    ids = set(range(1, len(kandidaten) + 1))
    verdict = _yesno_schema(_json_file(path), ids)
    if verdict is None:
        status, verdict = await run_single_slot(
            ctx, "Anker-Beleg", key=f"blocks-{topic}-anker-beleg-{h}",
            prompt=_prompt("Blocks-Supplement-Beleg", topic=topic, proposals="\n\n".join(lines),
                           extra=_extra(flow.state.get("instructions", ""))),
            role="judge", capabilities="none",
            payload=lambda result, p2=path, i=ids: _sink_json(result, p2, lambda d: _yesno_schema(d, i)),
            timeout=_timeout("selection_mapping", len(kandidaten)))
        if status != OK or not isinstance(verdict, dict):
            verdict = {}  # fail-open
    return hart_nein | {cid for k, (cid, _p) in enumerate(kandidaten, 1)
                        if str(verdict.get(k, "ja")) == "nein"}


async def _proc_consensus_gate(ctx: GenContext, flow: Flow, cards):
    """Code gate: reader union ≥2 (or supplement) passes; single finds → clarify.
    Reference-titled consensus clusters also go to clarify (majority quorum + rename).
    uni/projekt: quorum titles without ANY corpus anchor face the evidence judge —
    reader co-hallucination of famous canon beats the quorum otherwise."""
    topic = flow.topic
    moves = []
    anker_kandidaten: list[tuple[str, dict]] = []
    folder = source_folder(topic)
    ctoks = None
    if folder is not None:
        ctoks = flow.state.get("korpus_tokens")
        if ctoks is None:
            ctoks = flow.state["korpus_tokens"] = await asyncio.to_thread(_korpus_tokens, folder)
    for c in cards:
        cid = c["card_id"]
        rows = await _member_rows(topic, cid)
        if not rows:
            moves.append((cid, "done_cluster"))
            continue
        readers = set().union(*[set(r.get("readers") or []) for r in rows])
        supplement = any(r.get("supplement") for r in rows)
        rep = _rep(rows)
        p = c["payload"]
        p.update(title=rep["title"], description=rep["description"], readers=sorted(readers),
                 supplement=supplement, n_size=len(readers))  # LPT: big evidence first
        if supplement or len(readers) >= 2:
            if _is_reference(rep["title"]) and not supplement:
                p["quorum"] = "majority"  # consensus reference title: rename/exam, not the hard bar
                moves.append((cid, "clarify"))
            elif ctoks and not supplement and not _hat_anker(rep["title"], ctoks):
                anker_kandidaten.append((cid, p))
            else:
                moves.append((cid, "naming"))
        else:
            moves.append((cid, "clarify"))
        await db.kanban_set_payload(topic, BOARD, cid, p)
    if anker_kandidaten:
        weg = await _anker_beleg(ctx, flow, anker_kandidaten)
        for cid, p in anker_kandidaten:
            if cid in weg:
                p["reason"] = "kein-beleg"
                await db.kanban_set_payload(topic, BOARD, cid, p)
                moves.append((cid, "rejected"))
            else:
                moves.append((cid, "naming"))
        if weg:
            _log(topic, f"Anker-Beleg: {len(weg)} Quorum-Titel ohne Materialbeleg verworfen")
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _proc_clarify(ctx: GenContext, flow: Flow, cards):
    """Single-reader finds: deterministic pre-reject, then a 3-judge panel (Blocks-Klaerung).
    Quorum: UNANIMITY for single-reader clusters (origin-split C1), majority for reference-titled
    consensus clusters. Kept reference titles get the panel's rename."""
    topic = flow.topic
    moves: list[tuple[str, str]] = []
    pending = []
    for c in cards:
        t = c["payload"].get("title", "")
        d = c["payload"].get("description", "")
        if (_is_artifact(t) or _FILTER_NOTATION.search(t)) and not _is_named_statement(t, d):
            c["payload"]["reason"] = "pre-reject"
            await db.kanban_set_payload(topic, BOARD, c["card_id"], c["payload"])
            moves.append((c["card_id"], "rejected"))
        else:
            pending.append(c)
    if pending:
        h = _h(*[c["card_id"] for c in pending])
        rest = "\n".join(f"- {_t_text(c['payload'])}" for c in pending)
        judges = [(j, flow.work_dir / f"clarify-{h}-j{j}.json")
                  for j in range(1, CONSOLIDATION_PANEL + 1)]

        async def _judge(j, path):
            if _runde_schema(_json_file(path)) is not None:
                return  # resume: keep a valid judge file
            await run_single_slot(
                ctx, f"Klärung j{j}", key=f"blocks-{topic}-clarify-{h}-j{j}",
                prompt=_prompt("Blocks-Klaerung", topic=topic, rest=rest,
                               final="\n- Entscheide JEDEN Eintrag. `rest` MUSS leer sein.",
                               out_path=path),
                role="judge", capabilities="files",
                payload=lambda result, p=path: _runde_schema(_json_file(p)),
                timeout=_timeout("selection_mapping", len(pending)))

        await asyncio.gather(*[_judge(j, p) for j, p in judges], return_exceptions=True)
        if ctx.is_cancelled():
            return
        outs, raws = [], []
        for _, path in judges:
            raw = _json_file(path)
            r = _runde_schema(raw)
            if r is not None:
                outs.append(r)
                raws.append(raw)
        if not outs:
            raise RuntimeError("Klärung: kein Judge lieferte ein Ergebnis")
        votes: dict[str, int] = {}
        for accepted, _ in outs:
            for nt in {_norm_title(_title(t)) for t in accepted}:
                votes[nt] = votes.get(nt, 0) + 1
        renames: dict[str, dict[str, int]] = {}
        for raw in raws:
            if isinstance(raw, dict) and isinstance(raw.get("rename"), dict):
                for old, new in raw["rename"].items():
                    new = str(new).strip()
                    if new:
                        renames.setdefault(_norm_title(str(old)), {}).setdefault(new, 0)
                        renames[_norm_title(str(old))][new] += 1
        for c in pending:
            p = c["payload"]
            norm = _norm_title(p.get("title", ""))
            v = votes.get(norm, 0)
            majority = p.get("quorum") == "majority"
            accept = (v * 2 >= len(outs)) if majority else (v >= len(outs))
            if not accept:
                p.update(reason="failed-quorum", votes=v, judges=len(outs))
                await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
                moves.append((c["card_id"], "rejected"))
                continue
            if _is_reference(p.get("title", "")) and (sug := renames.get(norm)):
                best = max(sug, key=lambda k: (sug[k], len(k)))
                if not _is_reference(best):
                    p.update(title=best, renamed=True)
            await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
            moves.append((c["card_id"], "naming"))
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _choose_title(ctx: GenContext, flow: Flow, cid: str, rows: list[dict],
                        template: str, current_title: str | None = None) -> tuple[str, str | None]:
    """Judge picks the best member title and MAY propose a short abstracted name — kept
    only when anchored (_name_verankert). → (member_norm, custom_title|None);
    ("", None) on cancel; {"ok":true} in the check keeps the current title."""
    topic = flow.topic
    h = _h(*[r["norm"] for r in rows], template)
    path = flow.work_dir / f"naming-{cid}-{h}.json"
    verdict = _naming_schema(_json_file(path), len(rows))
    if verdict is None:
        lines = "\n".join(f"{k + 1}. {_t_text(r)}" for k, r in enumerate(rows))
        kw = dict(topic=topic, members=lines)
        if current_title is not None:
            kw["current_title"] = current_title
        status, verdict = await run_single_slot(
            ctx, f"Naming {cid}", key=f"blocks-{topic}-naming-{cid}-{h}",
            prompt=_prompt(template, **kw), role="judge", capabilities="none",
            payload=lambda result, p=path, n=len(rows): _sink_json(result, p, lambda d: _naming_schema(d, n)),
            timeout=_timeout("selection_mapping", len(rows)))
        if status == CANCELLED:
            return "", None
        if status == FAILED:
            verdict = None
    if verdict is None:
        rep = _rep(rows)
        w = _norm_title(rep["title"])
        norms = [r["norm"] for r in rows]
        return (w if w in norms else norms[0]), None
    best, name, ok = verdict
    if ok:  # check judge confirms the current (possibly custom) title: change nothing
        return "", (current_title or "")
    custom = None
    if name:
        ctoks = flow.state.get("korpus_tokens")
        if _name_verankert(name, rows, ctoks):
            custom = clean_title(name)
    return rows[best - 1]["norm"], custom


async def _proc_naming(ctx: GenContext, flow: Flow, cards):
    results = await asyncio.gather(*[_name_one(ctx, flow, c) for c in cards], return_exceptions=True)
    errs = [r for r in results if isinstance(r, Exception)]
    if errs:
        raise errs[0]
    flow.wake.set()


async def _name_one(ctx: GenContext, flow: Flow, c):
    topic = flow.topic
    cid = c["card_id"]
    p = c["payload"]
    rows = await _member_rows(topic, cid)
    if len(rows) > 1 and not p.get("renamed"):
        winner, custom = await _choose_title(ctx, flow, cid, rows, "Blocks-Naming")
        if not winner and custom is None:  # cancelled
            return
        if winner:
            p["main_norm"] = winner
            w = next(r for r in rows if r["norm"] == winner)
            p["title"] = custom or w["title"]
            p["description"] = w.get("description") or p.get("description", "")
            await db.kanban_set_payload(topic, BOARD, cid, p)
    await _saniere_one(ctx, flow, cid, p)
    await db.kanban_advance(topic, BOARD, cid, "naming_check")


async def _saniere_one(ctx: GenContext, flow: Flow, cid: str, p: dict) -> None:
    """Fremd-/Hygiene-Sanierung am Entstehungsort (nur Korpus-Quellen): Titel ohne
    Korpus-Anker auf die Oberflächenform des Materials umschreiben, leere Beschreibung
    belegt nachfassen. Läuft NACH der Member-Wahl auf dem finalen Payload — greift damit
    auch für Singleton-Cluster und unverankerte Naming-Gewinner. Vorschläge werden
    deterministisch validiert (_hat_anker, dieselbe Messlatte wie QA-fremd); fail-open:
    unverankerter Vorschlag oder Judge-Ausfall lässt die Karte unverändert weiter."""
    from qa import _distinctive
    topic = flow.topic
    folder = source_folder(topic)
    if folder is None:
        return  # thema: kein Korpus, keine Oberflächenform zum Verankern
    ctoks = flow.state.get("korpus_tokens")
    if ctoks is None:  # Resume: das Konsens-Gate dieses Flows lief nie
        ctoks = flow.state["korpus_tokens"] = await asyncio.to_thread(_korpus_tokens, folder)
    if not _sanierung_noetig(p, ctoks):
        return
    titel, beschr = p.get("title", ""), (p.get("description") or "").strip()
    toks = _distinctive(titel) | _distinctive(beschr)
    ev = _evidence_pack(folder, p.get("sources") or None, [" ".join(sorted(toks))],
                        budget=4000) if toks else ""
    if not ev:
        return  # ohne Auszüge kein Anker-Rewrite — Anker-Beleg hat Echtheit schon entschieden
    h = _h(titel, beschr, "sanierung")
    path = flow.work_dir / f"sanierung-{cid}-{h}.json"
    verdict = _sanierung_schema(_json_file(path))
    if verdict is None:
        status, verdict = await run_single_slot(
            ctx, f"Sanierung {cid}", key=f"blocks-{topic}-sanierung-{cid}-{h}",
            prompt=_prompt("Blocks-Sanierung", topic=topic, title=titel,
                           description=beschr or "(leer)", excerpts=ev),
            role="judge", capabilities="none",
            payload=lambda result, p2=path: _sink_json(result, p2, _sanierung_schema),
            timeout=_timeout("selection_mapping", 1))
        if status != OK or verdict is None:
            return
    neu_titel, neu_beschr = verdict
    changed = False
    if (neu_titel and _norm_title(neu_titel) != _norm_title(titel)
            and _hat_anker(neu_titel, ctoks)):
        p["title"] = clean_title(neu_titel)
        changed = True
    if not beschr and neu_beschr:
        p["description"] = neu_beschr
        changed = True
    if changed:
        await db.kanban_set_payload(topic, BOARD, cid, p)


async def _proc_naming_check(ctx: GenContext, flow: Flow, cards):
    results = await asyncio.gather(*[_namecheck_one(ctx, flow, c) for c in cards], return_exceptions=True)
    errs = [r for r in results if isinstance(r, Exception)]
    if errs:
        raise errs[0]
    flow.wake.set()


async def _namecheck_one(ctx: GenContext, flow: Flow, c):
    """Second judge verifies the title choice, then spawns the block card."""
    topic = flow.topic
    cid = c["card_id"]
    p = c["payload"]
    rows = await _member_rows(topic, cid)
    if len(rows) > 1 and not p.get("renamed"):
        winner, custom = await _choose_title(ctx, flow, cid, rows, "Blocks-Naming-Check",
                                             current_title=p.get("title", ""))
        if not winner and custom is None:  # cancelled
            return
        if winner:  # ok-verdict ("" + custom) keeps title AND description untouched
            w = next(r for r in rows if r["norm"] == winner)
            p["title"] = custom or w["title"]
            p["description"] = w.get("description") or p.get("description", "")
    readers = sorted(set().union(*[set(r.get("readers") or []) for r in rows])) if rows else []
    sources = sorted(set().union(*[set(r.get("sources") or []) for r in rows])) if rows else []
    await db.kanban_upsert_card(topic, BOARD, f"b-{cid}", "block", "fragment_filter", {
        "title": p.get("title", ""), "description": p.get("description", ""),
        "readers": readers, "sources": sources, "n_size": len(readers),
        "supplement": bool(p.get("supplement")), "cluster": cid,
    })
    await db.kanban_advance(topic, BOARD, cid, "done_cluster")


# ── Block barriers ─────────────────────────────────────────────────────────────────
async def _context_blocks(topic: str, exclude: set[str]) -> list[dict]:
    """Blocks already past the filter (grouping/gap_check/done_block) — context for the
    relational judges of a LATER pass (supplement feedback), never demotable themselves."""
    out = []
    for r in await db.kanban_cards(topic, board=BOARD, kind="block"):
        if r["card_id"] in exclude or r["stage"] not in ("grouping", "gap_check", "done", "done_block"):
            continue
        p = r["payload"]
        out.append({"card_id": r["card_id"], "payload": p,
                    "title": p.get("title", ""), "description": p.get("description") or "",
                    "title_norm": _norm_title(p.get("title", ""))})
    return out


async def _proc_fragment_filter(ctx: GenContext, flow: Flow, cards):
    """BARRIER/drain — the global pass over ALL block cards in the stage:
    (a) deterministic re-merge across cluster borders (canonical key + title-cos ≥0.95,
        catches cap splits), complete-link + relation guard, survivorship keeps the champion;
    (b) the degrade pass from blocks.py: full list per judge, hard-drop double gate,
        containment demote, parentless-noise drop, recheck panel, statement-gate rescue,
        transitive root resolution. Fragments/drops → rejected (journal), survivors → grouping."""
    topic = flow.topic
    work_dir = flow.work_dir
    rows = [{"card_id": c["card_id"], "payload": c["payload"],
             "title": c["payload"].get("title", ""),
             "description": c["payload"].get("description") or "",
             "title_norm": _norm_title(c["payload"].get("title", ""))} for c in cards]
    moves: list[tuple[str, str]] = []

    # (a) global re-merge
    n = len(rows)
    if n >= 2:
        edges = []
        vt = await _vec_rows(flow, [r["title"] for r in rows])
        keys: dict[str, list[int]] = {}
        for i, r in enumerate(rows):
            if (k := _canonical_key(r["title"])):
                keys.setdefault(k, []).append(i)
        for grp in keys.values():
            for x in range(len(grp)):
                for y in range(x + 1, len(grp)):
                    a, b = grp[x], grp[y]
                    if not _relation_conflict(rows[a]["title"], rows[b]["title"]):
                        edges.append((a, b))
        if vt is not None:
            sims_t = vt @ vt.T
            for i in range(n):
                for j in range(i + 1, n):
                    if float(sims_t[i][j]) >= DEDUP_TITLE_AUTO and not _relation_conflict(
                            rows[i]["title"], rows[j]["title"]):
                        edges.append((i, j))
        merged = 0
        for g in _cliques(n, edges):
            rep = max(g, key=lambda k: (-_aspect_marker(rows[k]["title"]),
                                        len(rows[k]["description"]), len(rows[k]["title"]), -k))
            rp = rows[rep]["payload"]
            for k in g:
                if k == rep:
                    continue
                lp = rows[k]["payload"]
                rp["readers"] = sorted(set(rp.get("readers") or []) | set(lp.get("readers") or []))
                rp["sources"] = sorted(set(rp.get("sources") or []) | set(lp.get("sources") or []))
                lp.update(reason="merged", merged_into=rows[rep]["title"])
                await db.kanban_set_payload(topic, BOARD, rows[k]["card_id"], lp)
                moves.append((rows[k]["card_id"], "grouped"))
                merged += 1
            await db.kanban_set_payload(topic, BOARD, rows[rep]["card_id"], rp)
        if merged:
            dropped = {cid for cid, _ in moves}
            rows = [r for r in rows if r["card_id"] not in dropped]
            _log(topic, f"Fragment-Filter: {merged} Cap-Split-Dublette(n) re-merged")

    # (b) degrade pass — full list = demotable stage rows + already-confirmed context
    context = await _context_blocks(topic, exclude={r["card_id"] for r in rows})
    allrows = rows + context
    n_dem, n_all = len(rows), len(allrows)
    if n_dem:
        def _fline(i):
            r = allrows[i - 1]
            mark = "⚠ " if i <= n_dem and _filter_suspect(r) else ""
            tail = "" if i <= n_dem else " (bereits bestätigt)"
            d = r.get("description")
            return (f"{i}. {mark}{r['title']} — {d}{tail}" if d else f"{i}. {mark}{r['title']}{tail}")

        full_list = "\n".join(_fline(i) for i in range(1, n_all + 1))
        h = _h(*[r["card_id"] for r in rows])
        chunks = [list(range(i, min(i + FILTER_CHUNK, n_dem + 1))) for i in range(1, n_dem + 1, FILTER_CHUNK)]

        # ONE wave: all chunk judges in parallel (verdicts are independent; the voting/
        # containment evaluation below reads the files strictly afterwards).
        async def _judge_chunk(ci, numbers):
            path = work_dir / f"filter-{h}-c{ci}.json"
            if _filter_schema(_json_file(path)) is not None:
                return  # resume: keep a valid judge file
            status, _v = await run_single_slot(
                ctx, f"Fragment-Filter {ci}", key=f"blocks-{topic}-filter-{h}-c{ci}",
                prompt=_prompt("Blocks-Filter", topic=topic, list=full_list,
                               from_n=numbers[0], to_n=numbers[-1], out_path=path),
                role="judge", capabilities="files",
                payload=lambda result, p=path: _filter_schema(_json_file(p)),
                timeout=_timeout("selection_mapping", len(numbers)))
            if status == FAILED:
                raise RuntimeError(f"Fragment-Filter chunk {ci} ohne Ergebnis")

        results = await asyncio.gather(*[_judge_chunk(ci, nm) for ci, nm in enumerate(chunks)],
                                       return_exceptions=True)
        errs = [r for r in results if isinstance(r, Exception)]
        if errs:
            raise errs[0]
        if ctx.is_cancelled():
            return
        proposals: dict[int, int] = {}  # judge demotes are PROPOSALS — panel/containment confirm
        drops: set[int] = set()
        for ci, numbers in enumerate(chunks):
            raw = _json_file(work_dir / f"filter-{h}-c{ci}.json")
            verdict = _filter_schema(raw) or {}
            nset = set(numbers)
            for nr, parent in verdict.items():
                if 1 <= parent <= n_all and nr in nset:
                    proposals[nr] = parent
            for x in (raw.get("drop", []) if isinstance(raw, dict) else []):
                try:
                    dnr = int(x)
                except (ValueError, TypeError):
                    continue
                if dnr in nset:
                    drops.add(dnr)
        # hard-drop double gate
        honored = {nr for nr in drops if _is_artifact(allrows[nr - 1]["title"])}
        for nr in honored:
            proposals.pop(nr, None)
        # containment demote (deterministic — auto-confirms proposals whose parent name is
        # literally contained in the title, and keeps the legacy ⚠-only pass) + parentless noise
        norms = [(i, allrows[i - 1]["title_norm"]) for i in range(1, n_all + 1)]
        fragments: dict[int, int] = {}
        contained: set[int] = set()
        for i in range(1, n_dem + 1):
            if i in honored or (i not in proposals and not _filter_suspect(allrows[i - 1])):
                continue
            parent = _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i])
            if parent is not None and parent != i and parent not in honored:
                fragments[i] = parent
                contained.add(i)
                proposals.pop(i, None)
        for i in range(1, n_dem + 1):
            if i in fragments or i in honored or i in proposals or not _is_parentless_noise(allrows[i - 1]["title"]):
                continue
            if _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i]) is None:
                honored.add(i)
        # recheck panel = second opinion: unconfirmed judge proposals (WITHOUT the suggested
        # parent — no anchoring) plus still-⚠ survivors. Majority ≥2 demotes/drops; a proposal
        # the panel does not confirm survives. The panel is load-bearing now: fewer than 2
        # valid judge files per chunk is an error (backoff), not a silent keep.
        survivors = [i for i in range(1, n_dem + 1)
                     if i not in fragments and i not in honored
                     and (i in proposals or _filter_suspect(allrows[i - 1]))]
        overruled: list[int] = []
        if survivors:
            ph = _h(",".join(map(str, survivors)))  # panel-input hash: stale pre-change files never match
            rchunks = [survivors[k:k + FILTER_CHUNK] for k in range(0, len(survivors), FILTER_CHUNK)]

            async def _recheck_judge(ci, nums, j):
                path = work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json"
                if _filter_schema(_json_file(path)) is not None:
                    return  # resume
                await run_single_slot(
                    ctx, f"Filter-Recheck {ci}/{j}", key=f"blocks-{topic}-filter-recheck-{h}-{ph}-c{ci}-j{j}",
                    prompt=_prompt("Blocks-Filter-Recheck", topic=topic,
                                   survivors="\n".join(_fline(i) for i in nums),
                                   list=full_list, out_path=path),
                    role="judge", capabilities="files",
                    payload=lambda result, p=path: _filter_schema(_json_file(p)),
                    timeout=_timeout("selection_mapping", len(nums)))

            await asyncio.gather(*[
                _recheck_judge(ci, nums, j)
                for ci, nums in enumerate(rchunks) for j in range(1, FILTER_RECHECK_PANEL + 1)],
                return_exceptions=True)
            if ctx.is_cancelled():
                return
            for ci, nums in enumerate(rchunks):
                dem: dict[int, list[int]] = {}
                drp: dict[int, int] = {}
                nset = set(nums)
                valid = 0
                for j in range(1, FILTER_RECHECK_PANEL + 1):
                    raw = _json_file(work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json")
                    v = _filter_schema(raw)
                    if v is None:
                        continue
                    valid += 1
                    for nr, parent in v.items():
                        if nr in nset and 1 <= parent <= n_all and nr != parent:
                            dem.setdefault(nr, []).append(parent)
                    for x in (raw.get("drop", []) if isinstance(raw, dict) else []):
                        try:
                            dnr = int(x)
                        except (ValueError, TypeError):
                            continue
                        if dnr in nset:
                            drp[dnr] = drp.get(dnr, 0) + 1
                if valid < 2:
                    raise RuntimeError(f"Filter-Recheck chunk {ci}: nur {valid} Judge(s) mit Ergebnis")
                for nr in nums:
                    if nr in fragments or nr in honored:
                        continue
                    if drp.get(nr, 0) >= 2 and (_is_artifact(allrows[nr - 1]["title"])
                                                or _is_parentless_noise(allrows[nr - 1]["title"])):
                        honored.add(nr)
                    elif len(dem.get(nr, [])) >= 2:
                        fragments[nr] = max(set(dem[nr]), key=dem[nr].count)
                    elif nr in proposals:
                        overruled.append(nr)
        # embedding backstop: veto confirmed non-containment demotes whose direct title pair is
        # literally structureless (see FRAGMENT_MIN_COS) — applied BEFORE _root resolution.
        floor_veto: list[int] = []
        if (cand := [nr for nr in fragments if nr not in contained]):
            va = await _vec_rows(flow, [r["title"] for r in allrows])
            if va is not None:
                for nr in cand:
                    if float(va[nr - 1] @ va[fragments[nr] - 1]) < FRAGMENT_MIN_COS:
                        fragments.pop(nr)
                        floor_veto.append(nr)
        # statement-gate rescue (final override)
        def _protected(nr):
            return _is_named_statement(allrows[nr - 1]["title"], allrows[nr - 1]["description"])
        for nr in [nr for nr in list(fragments) if _protected(nr)]:
            fragments.pop(nr, None)
        honored -= {nr for nr in honored if _protected(nr)}
        # transitive resolution + moves
        journal = []
        for nr in sorted(fragments):
            root, cyclic = _root(nr, fragments)
            if cyclic or not (1 <= root <= n_all) or nr > n_dem:
                continue
            r = allrows[nr - 1]
            parent_norm = None if root in honored else allrows[root - 1]["title_norm"]
            r["payload"].update(reason="fragment" if parent_norm else "drop-collateral",
                                parent_norm=parent_norm)
            await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
            moves.append((r["card_id"], "rejected"))
            journal.append({"fragment": r["title"],
                            "eltern": allrows[root - 1]["title"] if parent_norm else None})
        for nr in sorted(honored):
            if nr > n_dem:
                continue
            r = allrows[nr - 1]
            r["payload"].update(reason="drop", parent_norm=None)
            await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
            moves.append((r["card_id"], "rejected"))
            journal.append({"fragment": r["title"], "eltern": None, "grund": "drop"})
        # one journal file per pass (h) — the supplement feedback pass must not overwrite
        # the main pass's journal (it is the evaluation instrument).
        atomic_write_json(work_dir / f"inventar-filter-{h}.json",
                          {"vorher": n_dem, "degradiert": len(journal), "fragments": journal,
                           "ueberstimmt": [allrows[nr - 1]["title"] for nr in overruled],
                           "floor_veto": [allrows[nr - 1]["title"] for nr in floor_veto]}, indent=1)
        _log(topic, f"Fragment-Filter: {n_dem} → {n_dem - len(journal)} (−{len(journal)})")
    demoted = {cid for cid, _ in moves}
    moves += [(r["card_id"], "dedup") for r in rows if r["card_id"] not in demoted]
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _proc_dedup(ctx: GenContext, flow: Flow, cards):
    """BARRIER/drain — global pair dedup over the NAMED blocks: filter's re-merge only
    catches key-exact/≥0.95 titles inside the stage; here embedding candidates (mean OR
    title cosine ≥ DEDUP_GLOBAL_FLOOR) + canonical-key blocking go to a TWO-judge panel
    (merge needs unanimity — single judges conflate variants with their base entity),
    auto edges (title ≥0.95, exact key) with relation guard, complete-link cliques merge
    into the champion (readers/sources union). The second wave (supplement) compares
    against the already-confirmed context blocks. Journal carries every pair verdict."""
    topic = flow.topic
    work_dir = flow.work_dir
    rows = [{"card_id": c["card_id"], "payload": c["payload"],
             "title": c["payload"].get("title", ""),
             "description": c["payload"].get("description") or ""} for c in cards]
    context = await _context_blocks(topic, exclude={r["card_id"] for r in rows})
    allrows = rows + context
    n_dem, n_all = len(rows), len(allrows)

    async def _pass_through():
        await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "grouping") for r in rows])
        flow.wake.set()

    if n_all < 2 or not await _emb_ok(flow):
        await _pass_through()
        return
    vf = await _vec_rows(flow, [_t_text(r) for r in allrows])
    # titles casefolded: ALL-CAPS variants ("VERTEX COVER" vs "Vertex Cover (VC)") tank the
    # raw title cosine below the candidate floor
    vt = await _vec_rows(flow, [r["title"].casefold() for r in allrows])
    if vf is None or vt is None:
        await _pass_through()
        return
    sims_title = vt @ vt.T
    sims = (vf @ vf.T + sims_title) / 2

    def _demotable_pair(a: int, b: int) -> bool:
        return a < n_dem or b < n_dem  # confirmed context blocks never merge among themselves

    pairs: set[tuple[int, int]] = set()
    for i in range(n_all):
        for j in range(i + 1, n_all):
            # title-only cosine as second candidate source: descriptions of the same
            # entity often stress different facets and dilute the mean below the floor
            if _demotable_pair(i, j) and (float(sims[i][j]) >= DEDUP_GLOBAL_FLOOR
                                          or float(sims_title[i][j]) >= DEDUP_GLOBAL_FLOOR):
                pairs.add((i, j))
    keys: dict[str, list[int]] = {}
    for i, r in enumerate(allrows):
        if (k := _canonical_key(r["title"])):
            keys.setdefault(k, []).append(i)
    for grp in keys.values():
        for x in range(len(grp)):
            for y in range(x + 1, len(grp)):
                if _demotable_pair(grp[x], grp[y]):
                    pairs.add((grp[x], grp[y]))
    # acronym↔expansion pairs ("SAT" vs "Satisfiability Problem (SAT)"): the title cosine
    # of short vs long form sits far below the floor — variant match makes them candidates,
    # the judge panel decides as usual (no auto-merge)
    idents = [{_norm_title(r["title"]), _canonical_key(r["title"])} - {""} for r in allrows]
    variants = [_title_variants(r["title"]) for r in allrows]
    for i in range(n_all):
        for j in range(i + 1, n_all):
            if _demotable_pair(i, j) and (variants[i] & idents[j] or variants[j] & idents[i]):
                pairs.add((i, j))
    ordered = sorted(pairs)
    h = _h(*[r["card_id"] for r in allrows])
    if not ordered:
        atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
                          {"vorher": n_dem, "paare": 0, "merged": []}, indent=1)
        await _pass_through()
        return
    chunks = [ordered[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(ordered), DEDUP_PAIRS_CHUNK)]

    async def _judge(ci, chunk, jj):
        path = work_dir / f"dedup-{h}-c{ci}-j{jj}.json"
        if _pairs_schema(_json_file(path)) is not None:
            return  # resume
        lines = "\n\n".join(f"{j + 1}.\nA: {_t_text(allrows[a])}\nB: {_t_text(allrows[b])}"
                            for j, (a, b) in enumerate(chunk))
        status, _v = await run_single_slot(
            ctx, f"Dedup {ci} j{jj}", key=f"blocks-{topic}-dedup-{h}-c{ci}-j{jj}",
            prompt=_prompt("Blocks-Dedup", topic=topic, pairs=lines, out_path=path),
            role="judge", capabilities="files",
            payload=lambda result, p=path: _pairs_schema(_json_file(p)),
            timeout=_timeout("selection_mapping", len(chunk)))
        if status == FAILED:
            raise RuntimeError(f"Dedup chunk {ci} j{jj} ohne Ergebnis")

    # two-judge panel per chunk (one wave): a merge needs UNANIMITY — the observed
    # failure mode is a single judge conflating a variant with its base entity
    results = await asyncio.gather(*[_judge(ci, c, jj) for ci, c in enumerate(chunks)
                                     for jj in (1, 2)], return_exceptions=True)
    errs = [r for r in results if isinstance(r, Exception)]
    if errs:
        raise errs[0]
    if ctx.is_cancelled():
        return
    edges: list[tuple[int, int]] = []
    detail: dict[tuple[int, int], str] = {}

    def _edge(a, b, kanal):
        if _relation_conflict(allrows[a]["title"], allrows[b]["title"]):
            detail[(a, b)] = "guard_veto"
        else:
            edges.append((a, b))
            detail[(a, b)] = kanal

    for ci, chunk in enumerate(chunks):
        v1 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j1.json")) or {}
        v2 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j2.json")) or {}
        for j, (a, b) in enumerate(chunk):
            ja1, ja2 = bool(v1.get(j + 1)), bool(v2.get(j + 1))
            if ja1 and ja2:
                _edge(a, b, "ja")
            else:
                detail[(a, b)] = "nein" if not (ja1 or ja2) else "uneinig"
    for a, b in ordered:  # auto recall net: near-identical titles merge without the judges
        if float(sims_title[a][b]) >= DEDUP_TITLE_AUTO:
            _edge(a, b, "auto_titel")
    for grp in keys.values():  # exact-canonical-key auto-merge
        for x in range(len(grp)):
            for y in range(x + 1, len(grp)):
                if _demotable_pair(grp[x], grp[y]):
                    _edge(grp[x], grp[y], "auto_key")
    moves: list[tuple[str, str]] = []
    journal: list[dict] = []
    gone: set[int] = set()
    for g in _cliques(n_all, edges):
        ctx_members = [k for k in g if k >= n_dem]
        if ctx_members:
            rep = ctx_members[0]  # confirmed block always wins
        else:
            rep = max(g, key=lambda k: (-_aspect_marker(allrows[k]["title"]),
                                        len(allrows[k]["description"]), len(allrows[k]["title"]), -k))
        rp = allrows[rep]["payload"]
        changed = False
        for k in g:
            if k == rep or k >= n_dem:
                continue  # context members stay untouched
            lp = allrows[k]["payload"]
            rp["readers"] = sorted(set(rp.get("readers") or []) | set(lp.get("readers") or []))
            rp["sources"] = sorted(set(rp.get("sources") or []) | set(lp.get("sources") or []))
            lp.update(reason="merged", merged_into=allrows[rep]["title"])
            await db.kanban_set_payload(topic, BOARD, allrows[k]["card_id"], lp)
            moves.append((allrows[k]["card_id"], "grouped"))
            gone.add(k)
            journal.append({"dublette": allrows[k]["title"], "in": allrows[rep]["title"]})
            changed = True
        if changed:
            await db.kanban_set_payload(topic, BOARD, allrows[rep]["card_id"], rp)
            if rep >= n_dem and rp.get("mirrored_norm"):  # mirror refresh (sources only)
                await db.upsert_block(topic, rp["mirrored_norm"], allrows[rep]["title"],
                                      allrows[rep]["description"], rp.get("sources") or [])
    moves += [(r["card_id"], "grouping") for i, r in enumerate(rows) if i not in gone]
    atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
                      {"vorher": n_dem, "paare": len(ordered), "merged": journal,
                       "paare_detail": [{"a": allrows[a]["title"], "b": allrows[b]["title"],
                                         "verdict": v} for (a, b), v in sorted(detail.items())]},
                      indent=1)
    if journal:
        _log(topic, f"Dedup: {len(journal)} Dublette(n) zusammengelegt")
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _proc_grouping(ctx: GenContext, flow: Flow, cards):
    """BARRIER/drain — umbrella grouping over the filter survivors: embedding sibling clusters
    (low floor, high recall) + top-down pass, one judge per cluster, type gate + min-cos backstop,
    reconcile pass, completion judge. Members → grouped; the umbrella becomes a new block card."""
    topic = flow.topic
    work_dir = flow.work_dir
    rows = [{"card_id": c["card_id"], "payload": c["payload"],
             "title": c["payload"].get("title", ""),
             "description": c["payload"].get("description") or "",
             "title_norm": _norm_title(c["payload"].get("title", ""))} for c in cards]
    n = len(rows)
    if not (BLOCKS_GRUPPIERUNG_AKTIV and await _emb_ok(flow)) or n < 3:
        await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "gap_check") for r in rows])
        flow.wake.set()
        return
    texts = [_t_text(r) for r in rows]
    vv = await _vec_rows(flow, texts)
    if vv is None:
        await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "gap_check") for r in rows])
        flow.wake.set()
        return
    sims = vv @ vv.T
    clusters = await asyncio.to_thread(embedding.capped_blocks, sims,
                                       EMBEDDING_SIBLING_FLOOR, EMBEDDING_SIBLING_CAP)
    multi = [c for c in clusters if len(c) > 1]
    all_ids = set(range(1, n + 1))
    full_list = "\n".join(f"{i}. {texts[i - 1]}" for i in range(1, n + 1))
    h = _h(*[r["card_id"] for r in rows])

    def _min_cos(idxs):
        if len(idxs) < 2:
            return 1.0
        return round(min(float(sims[i][j]) for a, i in enumerate(idxs) for j in idxs[a + 1:]), 3)

    async def _assess(tag, cand_text, count):
        path = work_dir / f"gruppierung-{h}-c{tag}.json"
        if _umbrella_schema(_json_file(path), all_ids) is None:
            status, _v = await run_single_slot(
                ctx, f"Gruppierung {tag}", key=f"blocks-{topic}-gruppierung-{h}-c{tag}",
                prompt=_prompt("Blocks-Gruppierung", topic=topic, candidates=cand_text,
                               list=full_list, out_path=path),
                role="judge", capabilities="files",
                payload=lambda result, p=path: _umbrella_schema(_json_file(p), all_ids),
                timeout=_timeout("research_mapping", count))
            if status == CANCELLED:
                return None
        return path

    # ONE wave: TOP + all cluster judges in parallel. The used-ties precedence lives in the
    # ORDER of `sources` (TOP first), not in execution order — a failed judge simply leaves
    # no valid file and is skipped in the collection below (legacy semantics).
    jobs = [("TOP", "Scan the ENTIRE block list below and propose EVERY genuine umbrella "
             "you find — do not restrict yourself to any subset.", n)]
    jobs += [(str(ci), "\n".join(f"{g + 1}. {texts[g]}" for g in cluster), len(cluster))
             for ci, cluster in enumerate(multi)]
    await asyncio.gather(*[_assess(tag, cand, count) for tag, cand, count in jobs],
                         return_exceptions=True)
    if ctx.is_cancelled():
        return
    sources = [work_dir / f"gruppierung-{h}-c{tag}.json" for tag, _, _ in jobs]  # TOP first → wins used-ties

    existing_norms = {r["title_norm"] for r in rows}
    for other in await db.kanban_cards(topic, board=BOARD, kind="block"):
        existing_norms.add(_norm_title(other["payload"].get("title", "")))
    used: set[int] = set()
    seen_norm = set(existing_norms)
    chosen, skipped = [], []
    for src in sources:
        for title, desc, members in (_umbrella_schema(_json_file(src), all_ids) or []):
            import re as _re
            title = _re.sub(r"\s+[—–]\s+", ": ", title).strip()
            members = [m for m in members if m not in used]
            if len(members) < 2:
                continue
            mrows = [rows[m - 1] for m in members]
            if any(_GROUP_STANDALONE.search(r["title"]) for r in mrows):
                skipped.append({"umbrella": title, "grund": "type-gate",
                                "mitglieder": [r["title"] for r in mrows]})
                continue
            mc = _min_cos([m - 1 for m in members])
            if mc < GROUP_MIN_COS_FLOOR:
                skipped.append({"umbrella": title, "grund": "min-cos", "min_cos": mc,
                                "mitglieder": [r["title"] for r in mrows]})
                continue
            unorm = _norm_title(title)
            member_norms = {r["title_norm"] for r in mrows}
            if unorm not in member_norms and unorm in seen_norm:
                skipped.append({"umbrella": title, "grund": "title-collision",
                                "mitglieder": [r["title"] for r in mrows]})
                continue
            used.update(members)
            seen_norm.add(unorm)
            chosen.append({"umbrella": title, "description": desc, "min_cos": mc, "members": members})
    # reconcile: same parent proposed twice under different titles → union
    if len(chosen) >= 2:
        uv = await _vec_rows(flow, [f"{c['umbrella']} — {c['description']}" for c in chosen])
        if uv is not None:
            u_sims = uv @ uv.T
            parent = list(range(len(chosen)))
            for i in range(len(chosen)):
                for j in range(i + 1, len(chosen)):
                    if float(u_sims[i][j]) >= GROUP_RECONCILE_FLOOR:
                        embedding._union(parent, i, j)
            comp: dict[int, list[int]] = {}
            for i in range(len(chosen)):
                comp.setdefault(embedding._find(parent, i), []).append(i)
            merged = []
            for grp in comp.values():
                if len(grp) == 1:
                    merged.append(chosen[grp[0]])
                    continue
                rep = max(grp, key=lambda gi: len(chosen[gi]["members"]))
                members = list(dict.fromkeys(m for gi in grp for m in chosen[gi]["members"]))
                merged.append({**chosen[rep],
                               "description": " · ".join(chosen[gi]["description"] for gi in grp),
                               "members": members})
            chosen = merged
    # completion: absorb leftover standalone parts (per-member type-guard veto)
    leftover = sorted(all_ids - used)
    if chosen and leftover:
        cp = work_dir / f"gruppierung-completion-{h}.json"
        add = _completion_schema(_json_file(cp), len(chosen), set(leftover))
        if add is None:
            anchors = "\n".join(
                f"UMBRELLA {k}: {c['umbrella']} — {c['description']}\n  bereits: "
                + ", ".join(rows[m - 1]["title"] for m in c["members"]) for k, c in enumerate(chosen))
            rest = "\n".join(f"{i}. {texts[i - 1]}" for i in leftover)
            status, add = await run_single_slot(
                ctx, "Gruppierung completion", key=f"blocks-{topic}-gruppierung-completion-{h}",
                prompt=_prompt("Blocks-Gruppierung-Completion", topic=topic, umbrellas=anchors,
                               rest=rest, out_path=cp),
                role="judge", capabilities="files",
                payload=lambda result, p=cp: _completion_schema(_json_file(p), len(chosen), set(leftover)),
                timeout=_timeout("research_mapping", len(leftover)))
            if status == CANCELLED:
                return
            if status != OK:
                add = []
        for k, new_members in (add or []):
            for m in new_members:
                if m in used or not (1 <= m <= n) or _GROUP_STANDALONE.search(rows[m - 1]["title"]):
                    continue
                used.add(m)
                chosen[k]["members"].append(m)
    # apply: umbrella card + member moves
    moves = []
    for c in chosen:
        mrows = [rows[m - 1] for m in c["members"]]
        readers = sorted(set().union(*[set(r["payload"].get("readers") or []) for r in mrows]))
        srcs = sorted(set().union(*[set(r["payload"].get("sources") or []) for r in mrows]))
        await db.kanban_upsert_card(topic, BOARD, f"b-u-{uuid.uuid4().hex[:8]}", "block", "gap_check", {
            "title": clean_title(c["umbrella"]), "description": c["description"],
            "readers": readers, "sources": srcs, "umbrella": True,
            "children": [r["title"] for r in mrows],
        })
        for r in mrows:
            r["payload"].update(reason="umbrella", merged_into=c["umbrella"])
            await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
            moves.append((r["card_id"], "grouped"))
    grouped = {cid for cid, _ in moves}
    moves += [(r["card_id"], "gap_check") for r in rows if r["card_id"] not in grouped]
    atomic_write_json(work_dir / "inventar-gruppierung.json",
                      {"vorher": n, "umbrellas": [{"umbrella": c["umbrella"],
                                                   "mitglieder": [rows[m - 1]["title"] for m in c["members"]]}
                                                  for c in chosen], "skipped": skipped}, indent=1)
    if chosen:
        _log(topic, f"Gruppierung: {len(chosen)} Umbrella(s), −{len(grouped)} Blöcke")
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


async def _proc_gap_check(ctx: GenContext, flow: Flow, cards):
    """BARRIER/drain — the finished blocks advance to `done` IMMEDIATELY (board 2 starts
    while the supplement searches). ONE supplement round runs as an async PRODUCER: the
    producer count keeps the flow alive and the research_done gates closed, so the new
    titles feed back through every gate in a clean second pass. Failure never fatal."""
    topic = flow.topic
    if not flow.state.get("supplement_done"):
        flow.state["supplement_done"] = True  # one round (loop stop), set before the task
        titles = [c["payload"].get("title", "") for c in cards]
        titles += [b["title"] for b in await _context_blocks(topic, exclude=set())]
        flow.add_producer()  # SYNC before create_task — gates/exit must see the producer

        async def _run():
            try:
                await _supplement_producer(ctx, flow, sorted({t for t in titles if t}))
            except Exception:
                log.exception("[%s] supplement failed", topic)
            finally:
                flow.done_producer()

        asyncio.create_task(_run())
    await db.kanban_advance_many(topic, BOARD, [(c["card_id"], "done") for c in cards])
    flow.wake.set()


async def _supplement_beleg(ctx: GenContext, flow: Flow, supplements: list) -> list:
    """Evidence gate for supplement proposals: keyword excerpts per proposal, ONE no-tool
    judge marks material coverage (ja/nein). Proposals without any matching excerpt drop
    immediately; a failed gate keeps nothing (creep is costlier than a lost bonus round)."""
    topic = flow.topic
    folder = source_folder(topic)
    packs = [(t, d, _evidence_pack(folder, None, [t], budget=6000)) for t, d in supplements]
    cands = [(t, d, ev) for t, d, ev in packs if ev]
    kept: list = []
    if cands:
        path = flow.work_dir / "supplement-beleg.json"
        ids = set(range(1, len(cands) + 1))
        verdict = _yesno_schema(_json_file(path), ids)
        if verdict is None:
            lines = "\n\n".join(f"{k}. {t} — {d}\nAUSZÜGE:\n{ev}"
                                for k, (t, d, ev) in enumerate(cands, 1))
            status, verdict = await run_single_slot(
                ctx, "Supplement-Beleg", key=f"blocks-{topic}-supplement-beleg",
                prompt=_prompt("Blocks-Supplement-Beleg", topic=topic, proposals=lines,
                               extra=_extra(flow.state.get("instructions", ""))),
                role="judge", capabilities="none",
                payload=lambda result, p=path, i=ids: _sink_json(result, p, lambda d2: _yesno_schema(d2, i)),
                timeout=_timeout("selection_mapping", len(cands)))
            if status != OK or not isinstance(verdict, dict):
                verdict = {}
        kept = [(t, d) for k, (t, d, ev) in enumerate(cands, 1) if str(verdict.get(k, "nein")) == "ja"]
    if len(kept) < len(supplements):
        _log(topic, f"Supplement: {len(supplements) - len(kept)}/{len(supplements)} ohne Materialbeleg verworfen")
    return kept


async def _supplement_producer(ctx: GenContext, flow: Flow, titles: list[str]):
    """One web agent proposes canonically missing blocks → new title cards (reader
    'supplement' skips only the ≥2 consensus bar, every other gate applies)."""
    topic = flow.topic
    path = flow.work_dir / "supplement.json"
    supplements = _supplement_schema(_json_file(path))
    if supplements is None:
        # Source-bound topics (uni/projekt/link): the MATERIAL defines the scope — the agent
        # compares inventory vs. material (files, no web). Only pure "thema" topics research
        # the canon on the web (measured: the web agent proposed 22 textbook blocks the
        # script never treats, all discarded by the Beleg gate — 7 wasted minutes).
        folder = source_folder(topic)
        template, caps = ("Blocks-Supplement-Material", "files") if folder else ("Blocks-Supplement", "full")
        status, supplements = await run_single_slot(
            ctx, "Supplement", key=f"blocks-{topic}-supplement",
            prompt=_prompt(template, topic=topic, project=folder,
                           blocks="\n".join(f"- {t}" for t in titles),
                           out_path=path, extra=_extra(flow.state.get("instructions", ""))),
            role="quick", capabilities=caps,
            payload=lambda result, p=path: _supplement_schema(_json_file(p)),
            timeout=_timeout("ergaenzung"))
        if status == CANCELLED:
            flow.state["supplement_done"] = False  # in-memory only; resume re-derives from the file
            return
        if status != OK:
            _log(topic, "Supplement fehlgeschlagen — übersprungen (optional)")
            supplements = []
    # Material anchoring (uni/projekt): the web agent proposes canonical knowledge BLIND
    # to the source (measured: 22/107 aak blocks were textbook standard the script never
    # treats). Only proposals the material itself covers may enter the inventory.
    if supplements and source_folder(topic):
        supplements = await _supplement_beleg(ctx, flow, supplements)
    # Dead lineage: blocks demoted by the fragment filter (and their cluster + title cards)
    # must NOT dedup a supplement proposal — their content is gone. A hit on a dead title
    # REOPENS the lineage instead: the title card rejoins its cluster (live re-cluster) and
    # the respawned block gets a fresh fragment_filter pass. failed-quorum/pre-reject stay
    # in the dedup: those were rejected as non-blocks, not lost as content.
    dead_reasons = {"fragment", "drop-collateral", "drop"}
    cards = await db.kanban_cards(topic, board=BOARD)
    dead_clusters = {c["payload"].get("cluster") for c in cards
                     if c["kind"] == "block" and c["stage"] == "rejected"
                     and c["payload"].get("reason") in dead_reasons}
    dead_clusters.discard(None)
    membership = await db.kanban_membership(topic)
    dead_titles = {nm for nm, cid in membership.items() if cid in dead_clusters}

    def _is_dead(c) -> bool:
        if c["kind"] == "title":
            return c["card_id"] in dead_titles
        if c["kind"] == "cluster":
            return c["card_id"] in dead_clusters
        return c["stage"] == "rejected" and c["payload"].get("reason") in dead_reasons

    known_norms, known_keys = set(), set()
    dead_by_norm: dict[str, str] = {}  # title norm/key → requeue-able title card_id
    dead_by_key: dict[str, str] = {}
    for c in cards:
        tt = c["payload"].get("title", "")
        if not tt:
            continue
        if _is_dead(c):
            if c["kind"] == "title":
                dead_by_norm.setdefault(c["card_id"], c["card_id"])
                if (k := _canonical_key(tt)):
                    dead_by_key.setdefault(k, c["card_id"])
            continue
        known_norms.add(_norm_title(tt))
        if (k := _canonical_key(tt)):
            known_keys.add(k)
    new = reopened = 0
    for t, d in (supplements or []):
        t, d = clean_title(t), clean_title(d)
        norm = _norm_title(t)
        key = _canonical_key(t)
        if not norm or norm in known_norms or (key and key in known_keys):
            continue
        known_norms.add(norm)
        dead_id = dead_by_norm.get(norm) or (dead_by_key.get(key) if key else None)
        if dead_id:
            card = await db.kanban_get_card(topic, BOARD, dead_id)
            if card:
                card["payload"]["supplement"] = True
                await db.kanban_set_payload(topic, BOARD, dead_id, card["payload"])
                await db.kanban_advance(topic, BOARD, dead_id, "cluster")
                reopened += 1
                continue
        desc = f"{d} [Supplement]".strip()
        async with _ingest_lock:
            await db.kanban_add_title(topic, BOARD, norm, t, desc, "supplement", "supplement")
        card = await db.kanban_get_card(topic, BOARD, norm)
        if card:
            card["payload"]["supplement"] = True
            await db.kanban_set_payload(topic, BOARD, norm, card["payload"])
        new += 1
    if new or reopened:
        _log(topic, f"Supplement: {new} Block-Kandidat(en) → ingest, {reopened} wiedereröffnet")
        flow.wake.set()


async def _proc_done(ctx: GenContext, flow: Flow, cards):
    """Mirror finished blocks into the legacy `blocks` table (status consensus) — the
    interface everything downstream (guide, exam, frontend) already reads. Unique-title
    collisions get a numeric suffix; a re-run under a new name discards the old mirror."""
    topic = flow.topic
    mirrored: dict[str, str] = flow.state.setdefault("mirrored", {})
    moves = []
    for c in cards:
        p = c["payload"]
        base = p.get("title", "")
        if not base:
            moves.append((c["card_id"], "done_block"))
            continue
        title, norm, i = base, _norm_title(base), 2
        while mirrored.get(norm) not in (None, c["card_id"]):
            title = f"{base} ({i})"
            norm = _norm_title(title)
            i += 1
        old = p.get("mirrored_norm")
        if old and old != norm:
            await db.set_block_status(topic, old, "discarded")
        await db.upsert_block(topic, norm, title, p.get("description", ""), p.get("sources") or [])
        await db.set_block_status(topic, norm, "consensus",
                                  title=title, description=p.get("description", ""))
        mirrored[norm] = c["card_id"]
        p.update(mirrored_norm=norm, title=title)
        await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
        if (spawn := flow.state.get("spawn_artefact")):
            await spawn(c["card_id"], p)  # board 2 card per finished block
        moves.append((c["card_id"], "done_block"))
    await db.kanban_advance_many(topic, BOARD, moves)
    flow.wake.set()


# ── Orchestration ──────────────────────────────────────────────────────────────────
def inventory_stages(ctx: GenContext, flow: Flow) -> list[Stage]:
    research_done = lambda: flow.research_done  # noqa: E731
    return [
        Stage(BOARD, "ingest", lambda cs: _proc_ingest(flow, cs)),
        Stage(BOARD, "cluster", lambda cs: _proc_cluster(ctx, flow, cs), serial=True),
        Stage(BOARD, "pair_check", lambda cs: _proc_pair_check(ctx, flow, cs)),
        # gate (no barrier): hold clusters until ALL research producers are done — a title's
        # second reader may arrive minutes later, and the ≥2 vote must count it.
        Stage(BOARD, "consensus_gate", lambda cs: _proc_consensus_gate(ctx, flow, cs),
              gate=research_done),
        Stage(BOARD, "clarify", lambda cs: _proc_clarify(ctx, flow, cs)),
        Stage(BOARD, "naming", lambda cs: _proc_naming(ctx, flow, cs)),
        Stage(BOARD, "naming_check", lambda cs: _proc_naming_check(ctx, flow, cs)),
        Stage(BOARD, "fragment_filter", lambda cs: _proc_fragment_filter(ctx, flow, cs),
              barrier=True, drain=True, gate=research_done),
        Stage(BOARD, "dedup", lambda cs: _proc_dedup(ctx, flow, cs),
              barrier=True, drain=True, gate=research_done),
        Stage(BOARD, "grouping", lambda cs: _proc_grouping(ctx, flow, cs),
              barrier=True, drain=True, gate=research_done),
        Stage(BOARD, "gap_check", lambda cs: _proc_gap_check(ctx, flow, cs),
              barrier=True, drain=True, gate=research_done),
        Stage(BOARD, "done", lambda cs: _proc_done(ctx, flow, cs)),
    ]


async def _preload_state(flow: Flow):
    """Continue/resume: rebuild the in-memory caches from the persisted cards."""
    titles = await db.kanban_cards(flow.topic, board=BOARD, kind="title")
    known = [t for t in titles if t["stage"] != "ingest"]
    flow.state["nn"] = {"norms": [t["card_id"] for t in known],
                        "texts": [_t_text(t["payload"]) for t in known]}
    flow.state["mirrored"] = {
        c["payload"]["mirrored_norm"]: c["card_id"]
        for c in await db.kanban_cards(flow.topic, board=BOARD, stage="done_block")
        if c["payload"].get("mirrored_norm")}
    flow.state["supplement_done"] = _supplement_schema(
        _json_file(flow.work_dir / "supplement.json")) is not None


async def run_boards(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str,
                     research: bool = True, artefacts: bool = True, qa_force: bool = False) -> bool:
    """Run the inventory board (plus board 2 „Artefakte") until quiescence.
    research=False = Continue: drain the existing queue, search nothing new.
    qa_force=True overrides a failed QA gate (user clicked „Trotzdem fortsetzen")."""
    topic = ctx.topic
    flow = Flow(topic, files["arbeit"])
    flow.state["instructions"] = instructions
    flow.state["qa_force"] = qa_force
    run_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M')}-{uuid.uuid4().hex[:4]}"
    flow.state["run_id"] = run_id
    flow.state["run_started"] = datetime.now(timezone.utc).isoformat()
    db.set_current_run(topic, run_id)  # every event of this flow carries the run_id
    if q["type"] == "link" and folder:
        pages = await db.list_content(topic)
        flow.state["pages"] = pages or sorted(set(_crawl_index(folder).values()))
    await _preload_state(flow)
    stages = inventory_stages(ctx, flow)
    inv_names = [st.stage for st in stages]
    if artefacts:
        import board_artefacts  # lazy — board_artefacts imports blocks too
        flow.state["spawn_artefact"] = board_artefacts.make_spawner(topic, files)
        await board_artefacts.ensure_outline_card(topic)
        stages += board_artefacts.artefact_stages(ctx, flow, files, q, folder, instructions)
        if QA_GATE_NOTE > 0:
            # QA gate: board 2 waits until the inventory QA passed (or the user forces).
            # Costs pipelining (board 2 no longer starts per finished block) but saves
            # tokens on a bad foundation — the watcher below runs the QA and decides.
            sub = next(st for st in stages if st.stage == "subblocks")
            sub.gate = lambda: bool(flow.state.get("qa_ok") or flow.state.get("qa_force"))
    stages = chain_stages(stages)
    if artefacts:
        # Outline needs every block's TITLE + FACTS, nothing later: cut the post-facts
        # artefact stages from its barrier so it runs parallel to levels…finalize of the
        # slowest block (makespan tail). Inventory stages all stay — no late blocks.
        outline = next(s for s in stages if s.stage == "outline")
        outline.upstream = [u for u in outline.upstream if u not in
                            ("levels", "relevance", "question_pattern", "artefacts", "finalize")]
    producers = _build_producers(ctx, flow, q, folder, instructions) if research else []

    async def _as_producer(coro):
        try:
            await coro
        except Exception:
            log.exception("[%s] research producer failed", topic)
        finally:
            flow.done_producer()

    for _ in producers:
        flow.add_producer()
    flow.spawn_research = lambda: _research_once(
        ctx, flow, q, folder, instructions, f"x{flow.next_tag()}")
    # cancel hook: blocks.cancel_blocks flips is_cancelled; stop the flow with it
    stopper = asyncio.create_task(_stop_on_cancel(ctx, flow))
    watcher = (asyncio.create_task(_qa_gate_watch(ctx, flow, inv_names, set_p))
               if artefacts and QA_GATE_NOTE > 0 else None)
    try:
        try:
            await kanban.run_flow(flow, stages, [_as_producer(p) for p in producers], set_p)
        finally:
            stopper.cancel()
            if watcher:
                watcher.cancel()
        if ctx.is_cancelled():
            return False
        if flow.state.get("qa_paused"):
            return True  # kein Fehler: Flow hielt am QA-Gate, Karten warten in subblocks
        await _write_final(topic, files)
        await _write_run_summary(topic, flow)
        return True
    finally:
        # erst NACH der Abschluss-QA leeren: deren Judge-Events gehören zum Lauf —
        # vorher fielen sie ohne run_id aus jeder Run-Aggregation (Lauf 20260704-1452-b223)
        db.set_current_run(topic, None)


_QA_GATE_POLL = 2.0  # Sekunden zwischen Quiescence-Checks des QA-Wächters


async def _qa_gate_watch(ctx: GenContext, flow: Flow, inv_names: list[str], set_p):
    """Companion task: once the inventory is quiescent, run the QA once and decide —
    open the board-2 gate or pause the flow. Fail-OPEN on errors (QA is a helper,
    not a jailer); qa_force short-circuits to open."""
    topic = flow.topic
    try:
        while not flow.stop:
            if (flow.research_done and not flow.active_in(inv_names)
                    and await db.kanban_count(topic, inv_names) == 0):
                break
            await asyncio.sleep(_QA_GATE_POLL)
        if flow.stop or ctx.is_cancelled():
            return
        if flow.state.get("qa_force"):
            flow.state["qa_ok"] = True
            flow.wake.set()
            return
        set_p("QA prüft das Inventar…")
        import qa
        report = await qa.qa_report(topic, llm=QA_GATE_LLM)
        note = float(report["note"]) if report else 10.0
        flow.state["qa_note"] = note
        if report:
            try:  # Report-Persistenz ist Komfort — ein Schreibfehler darf das Gate nicht öffnen
                await qa.write_report(report)
            except Exception:
                log.exception("[%s] QA-Report schreiben fehlgeschlagen", topic)
        if note >= QA_GATE_NOTE:
            _log(topic, f"QA-Gate: Note {note} ≥ {QA_GATE_NOTE} — Board 2 startet")
            flow.state["qa_ok"] = True
        else:
            _log(topic, f"QA-Gate: Note {note} < {QA_GATE_NOTE} — pausiert (Continue erzwingt)")
            set_p(f"QA-Note {note} < {QA_GATE_NOTE} — pausiert")
            flow.state["qa_paused"] = True
            flow.stop = True
        flow.wake.set()
    except asyncio.CancelledError:
        raise
    except Exception:
        log.exception("[%s] QA-Gate fehlgeschlagen — Gate offen (fail-open)", topic)
        flow.state["qa_ok"] = True
        flow.wake.set()


async def _write_run_summary(topic: str, flow: Flow):
    """lauf-summary.json: the per-run numbers block QA diffs against. Never fatal."""
    try:
        run_id = flow.state.get("run_id", "")
        started = flow.state.get("run_started", "")
        finished = datetime.now(timezone.utc).isoformat()
        dauer = ""
        if started:
            dauer = round((datetime.fromisoformat(finished) - datetime.fromisoformat(started)).total_seconds() / 60, 1)
        summary = {"run_id": run_id, "topic": topic, "started": started, "finished": finished,
                   "dauer_min": dauer, "boards": await db.kanban_stage_counts(topic),
                   **await db.events_run_summary(topic, run_id)}
        try:  # Abschluss-QA MIT Judges: sub_dubletten/unechte werden beurteilt — erst damit
            import qa  # ist note_artefakte belastbar (Kandidatenliste allein zählt nicht)
            from config import ABSCHLUSS_QA_LLM  # Training schaltet die Judges ab (misst selbst)
            report = await qa.qa_report(topic, llm=bool(ABSCHLUSS_QA_LLM))
            if report:
                summary["note"] = report["note"]
                summary["note_artefakte"] = report.get("note_artefakte")
                summary["artefakte"] = report.get("artefakte", {})
                await qa.write_report(report)
        except Exception:
            log.exception("[%s] Abschluss-QA fehlgeschlagen", topic)
        atomic_write_json(flow.work_dir / "lauf-summary.json", summary, indent=1)
    except Exception:
        log.exception("[%s] lauf-summary fehlgeschlagen", topic)


async def _stop_on_cancel(ctx: GenContext, flow: Flow):
    while not flow.stop:
        if ctx.is_cancelled():
            flow.stop = True
            flow.wake.set()
            return
        await asyncio.sleep(0.3)


async def _write_final(topic: str, files: dict):
    """blocks.md from the finished cards (flow order — titles are already unique)."""
    done = await db.kanban_cards(topic, board=BOARD, stage="done_block")
    done.sort(key=lambda c: c["updated_at"])
    lines = [_line(i, c["payload"]) for i, c in enumerate(done, 1) if c["payload"].get("title")]
    if lines:
        atomic_write_text(files["final"], "\n".join(lines) + "\n")
    _log(topic, f"Kanban: fertig — {len(lines)} Blöcke")


# ── Board API (snapshot, reset, dead-letter) ───────────────────────────────────────
# (board, stage, label, kind) — display order of the live board. Terminal columns last.
COLUMNS = [
    ("inventory", "ingest", "Eingang", "title"),
    ("inventory", "cluster", "Cluster", "title"),
    ("inventory", "pair_check", "Paar-Check", "cluster"),
    ("inventory", "consensus_gate", "Konsens", "cluster"),
    ("inventory", "clarify", "Klärung", "cluster"),
    ("inventory", "naming", "Naming", "cluster"),
    ("inventory", "naming_check", "Naming-Check", "cluster"),
    ("inventory", "fragment_filter", "Fragment-Filter", "block"),
    ("inventory", "dedup", "Dubletten", "block"),
    ("inventory", "grouping", "Gruppierung", "block"),
    ("inventory", "gap_check", "Lücken-Check", "block"),
    ("inventory", "done", "Spiegeln", "block"),
    ("inventory", "done_block", "Fertig", "block"),
    ("inventory", "rejected", "Verworfen", None),
    ("inventory", "grouped", "Zusammengelegt", "block"),
    ("artefacts", "subblocks", "Subbausteine", "ablock"),
    ("artefacts", "facts", "Fakten", "ablock"),
    ("artefacts", "levels", "Stufen", "ablock"),
    ("artefacts", "relevance", "Relevanz", "ablock"),
    ("artefacts", "konsolidierung", "Konsolidierung", "ablock"),
    ("artefacts", "question_pattern", "Fragen", "ablock"),
    ("artefacts", "artefacts", "Lernkarten", "ablock"),
    ("artefacts", "finalize", "Zusammenführen", "ablock"),
    ("artefacts", "outline", "Gliederung", "outline"),
    ("artefacts", "done_artefact", "Fertig", "ablock"),
]

_TITLE_STAGES = ["ingest", "cluster"]
_CLUSTER_STAGES = ["pair_check", "consensus_gate", "clarify", "naming", "naming_check"]
_BLOCK_STAGES = ["fragment_filter", "dedup", "grouping", "gap_check", "done"]
_ART_STAGES = ["subblocks", "facts", "levels", "relevance", "konsolidierung", "question_pattern",
               "artefacts", "finalize", "outline"]
# where a requeued dead card restarts, by kind
_DEAD_RESTART = {"title": "cluster", "cluster": "pair_check", "block": "fragment_filter",
                 "ablock": "subblocks", "outline": "outline"}
_VERDICT_KEYS = ("reason", "votes", "judges", "merged_into", "parent_norm", "mirrored_norm")
DONE_ART = "done_artefact"


# Board-2 stage → its fine-step group in blocks.PHASEN (drives the per-card stepper).
_STAGE_PHASE = {"subblocks": "Subblocks", "facts": "Facts", "levels": "Levels",
                "relevance": "Relevance", "question_pattern": "Questions", "artefacts": "Artefacts"}
_PHASE_STEPS = {name: steps for name, steps in blocks.PHASEN}


def _card_view(r: dict, active: set[str], live_info: dict) -> dict:
    p = r["payload"]
    key = f"{r['board']}:{r['card_id']}"
    is_active = key in active
    info = r.get("last_error") or p.get("reason") or ""
    if p.get("merged_into"):
        info = f"→ {p['merged_into']}"
    elif p.get("parent_norm"):
        info = f"Fragment von: {p['parent_norm']}"
    out = {"title": p.get("title") or r["card_id"], "retries": r["retries"], "card_id": r["card_id"],
           "kind": r.get("kind", ""), "board": r.get("board", ""),
           "status": "error" if r["retries"] else ("active" if is_active else "open")}
    live = live_info.get(key)
    if is_active and live:  # live step message wins while the card is worked
        info = live["msg"] if isinstance(live, dict) else live  # dict since the stepper, str before
        step = live.get("step") if isinstance(live, dict) else ""
        steps = _PHASE_STEPS.get(_STAGE_PHASE.get(r.get("stage", ""), ""), ())
        if step in steps:  # phase stepper: which fine step of the card's stage runs (1-based)
            out.update(step_i=steps.index(step) + 1, step_n=len(steps), steps=list(steps))
    out["info"] = info
    return out


async def board_snapshot(topic: str, limit: int = 20) -> dict:
    """Live board: per column count + the newest cards (title, status, info)."""
    counts = await db.kanban_stage_counts(topic)
    flow = kanban.active_flows.get(topic)
    active = flow.active_cards if flow else set()
    live_info = flow.state.get("card_info", {}) if flow else {}
    columns = []
    for board, stage, label, _kind in COLUMNS:
        total = counts.get(board, {}).get(stage, 0)
        cards = ([_card_view(r, active, live_info)
                  for r in await db.kanban_stage_cards(topic, board, stage, limit)]
                 if total else [])
        columns.append({"board": board, "key": stage, "label": label, "total": total, "cards": cards})
    dead = [{"card_id": r["card_id"], "board": r["board"], "kind": r["kind"],
             "title": r["payload"].get("title") or r["card_id"], "error": r.get("last_error") or ""}
            for r in await db.kanban_dead(topic)]
    return {"columns": columns, "dead": dead,
            "done": counts.get("inventory", {}).get("done_block", 0),
            "qa": _qa_view(topic, counts, flow)}


def _qa_view(topic: str, counts: dict, flow) -> dict | None:
    """Latest QA report digest for the board header. `pausiert` = the gate stopped the
    flow (score below threshold, board-2 cards waiting, no flow running)."""
    # no inventory (deleted/never built) → no badge; the report files stay on purpose,
    # so the first run after a rebuild diffs against the old state
    if not counts.get("inventory", {}).get("done_block", 0) and not (
            flow and flow.state.get("qa_note") is not None):
        return None
    import qa
    tdir = qa.QA_DIR / topic
    # by mtime: a re-run overwrites the run-id-named file, which sorts before timestamp names.
    # guide-* reports are the guide_qa series — they must not shadow the inventory badge.
    reports = sorted((p for p in tdir.glob("*.json") if not p.name.startswith("guide-")),
                     key=lambda p: p.stat().st_mtime) if tdir.is_dir() else []
    if not reports:
        return None
    r = _json_file(reports[-1]) or {}
    note = r.get("note")
    if note is None:
        return None
    wartend = counts.get("artefacts", {}).get("subblocks", 0)
    pausiert = bool(note < QA_GATE_NOTE and wartend and flow is None)
    return {"note": note, "note_artefakte": r.get("note_artefakte"),
            "schwelle": QA_GATE_NOTE, "pausiert": pausiert,
            "quoten": r.get("quoten", {}),
            "befunde": (r.get("fremd", []) + r.get("unecht", []))[:6]}


async def _clean_artefact_state(topic: str, files: dict) -> None:
    """Artefact data is fully derived — wipe cards, DB tables and sidecar files."""
    await db.kanban_delete_cards(topic, "artefacts")
    await db.delete_subblocks(topic)
    await db.delete_question_pattern(topic)
    await db.delete_sub_artefakte(topic)
    await db.delete_outline(topic)
    for key in ("sub_roh", "sidecar", "facts", "question_pattern", "artefakte", "outline"):
        files[key].unlink(missing_ok=True)


async def reset_board_from_stage(topic: str, board: str, stage: str, files: dict) -> int:
    """Reset ab Spalte: cards from `stage` onward (incl. terminal rejected/grouped/dead)
    return to `stage`; downstream derived state is wiped. → moved card count.
    Only call while nothing is generating (the route guards)."""
    moved = 0

    async def _requeue(card: dict, to_stage: str):
        nonlocal moved
        p = {k: v for k, v in card["payload"].items() if k not in _VERDICT_KEYS}
        await db.kanban_set_payload(topic, card["board"], card["card_id"], p)
        await db.kanban_advance(topic, card["board"], card["card_id"], to_stage)
        moved += 1

    if board == "inventory" and stage in _TITLE_STAGES:
        # everything below titles is derived → full re-derive
        for r in await db.kanban_cards(topic, board="inventory"):
            if r["kind"] == "title":
                await _requeue(r, stage)
        await db.kanban_delete_cards(topic, "inventory", "cluster")
        await db.kanban_delete_cards(topic, "inventory", "block")
        dbc = await db.get_db()
        await dbc.execute("DELETE FROM kanban_members WHERE topic = ?", (topic,))
        await dbc.commit()
        await db.delete_blocks(topic)
        await _clean_artefact_state(topic, files)
    elif board == "inventory" and stage in _CLUSTER_STAGES:
        idx = _CLUSTER_STAGES.index(stage)
        later = set(_CLUSTER_STAGES[idx:]) | {"done_cluster", "rejected", "dead"}
        for r in await db.kanban_cards(topic, board="inventory", kind="cluster"):
            if r["stage"] in later:
                await _requeue(r, stage)
        await db.kanban_delete_cards(topic, "inventory", "block")
        await db.delete_blocks(topic)
        await _clean_artefact_state(topic, files)
    elif board == "inventory":
        idx = _BLOCK_STAGES.index(stage) if stage in _BLOCK_STAGES else 0
        later = set(_BLOCK_STAGES[idx:]) | {"done_block", "rejected", "grouped", "dead"}
        for r in await db.kanban_cards(topic, board="inventory", kind="block"):
            if r["stage"] in later:
                await _requeue(r, stage)
        await db.delete_blocks(topic)
        await _clean_artefact_state(topic, files)
    else:  # artefacts board
        idx = _ART_STAGES.index(stage) if stage in _ART_STAGES else 0
        later = set(_ART_STAGES[idx:]) | {DONE_ART, "dead"}
        for r in await db.kanban_cards(topic, board="artefacts"):
            if r["kind"] == "outline":
                await _requeue(r, "outline")
            elif r["stage"] in later:
                await _requeue(r, stage)
        if stage == "subblocks":  # full artefact re-derive → also the DB mirrors
            await db.delete_subblocks(topic)
            await db.delete_question_pattern(topic)
            await db.delete_sub_artefakte(topic)
            # global sidecar files and per-block resume slots: leftovers of the previous
            # derive would re-merge/resume into the fresh run — nothing here is reused
            for k in ("sidecar", "facts", "sub_roh", "question_pattern", "artefakte"):
                files[k].unlink(missing_ok=True)
            for d in files["arbeit"].glob("ab-*"):
                if d.is_dir():
                    shutil.rmtree(d, ignore_errors=True)
    await db.add_event(topic, "reset", key=f"{board}:from-{stage}", status=str(moved))
    return moved


async def restart_artefact_card(topic: str, card_id: str) -> bool:
    """Restart ONE artefacts card from `subblocks` — wipes only ITS derived DB rows
    (per-block work-dir slots overwrite themselves; finalize re-upserts later).
    Only call while nothing is generating (the route guards). → False if unknown."""
    card = await db.kanban_get_card(topic, "artefacts", card_id)
    if card is None or card["kind"] != "ablock":
        return False
    await db.delete_subblocks(topic, card_id)
    await db.delete_question_pattern(topic, card_id)
    await db.delete_sub_artefakte(topic, card_id)
    p = {k: v for k, v in card["payload"].items() if k in ("title", "description")}
    await db.kanban_set_payload(topic, "artefacts", card_id, p)
    await db.kanban_advance(topic, "artefacts", card_id, "subblocks")
    await db.add_event(topic, "reset", key=f"artefacts:{card_id}", status="card-restart")
    return True


async def requeue_dead(topic: str) -> int:
    """Dead-letter → restart stage by card kind (fresh retries). → requeued count."""
    n = 0
    for r in await db.kanban_dead(topic):
        stage = _DEAD_RESTART.get(r["kind"])
        if stage:
            await db.kanban_advance(topic, r["board"], r["card_id"], stage)
            n += 1
    return n


def add_research_agent(topic: str) -> bool:
    """Attach one more research agent to a live flow. Counted SYNCHRONOUSLY before the task
    (quiescence race). → True if a run was live to attach to."""
    flow = kanban.active_flows.get(topic)
    if flow is None or flow.stop or flow.spawn_research is None:
        return False
    flow.add_producer()

    async def _run():
        try:
            await flow.spawn_research()
        except Exception:
            log.exception("[%s] extra research failed", topic)
        finally:
            flow.done_producer()

    asyncio.create_task(_run())
    return True