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

import database as db
import embedding
import kanban
from kanban import Flow, Stage, chain_stages
import blocks
from blocks import (
    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,
)
from config import (
    BLOCKS_GRUPPIERUNG_AKTIV, EMBEDDING_AKTIV, EMBEDDING_BLOCK_CAP,
    EMBEDDING_SIBLING_CAP, EMBEDDING_SIBLING_FLOOR, 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, run_single_slot)
from textkit import _norm_title, _parse_selection, _title

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

BOARD = "inventory"
RESEARCH_RUNTIME = 900  # one research agent, one round — the tail ingests live while it writes
_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) -> int | None:
    """{"best": N} → 1-based member index in [1, count] · otherwise None."""
    if not isinstance(data, dict):
        return None
    try:
        n = int(data.get("best"))
    except (ValueError, TypeError):
        return None
    return n if 1 <= n <= count else None


# ── 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 = _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 = 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())


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)."""
    topic = flow.topic
    moves = []
    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)
        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"))
            else:
                moves.append((cid, "naming"))
        else:
            moves.append((cid, "clarify"))
        await db.kanban_set_payload(topic, BOARD, cid, p)
    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: int | None = None) -> str:
    """Judge picks the best member title (index). Parse-fail → survivorship fallback."""
    topic = flow.topic
    h = _h(*[r["norm"] for r in rows], template)
    path = flow.work_dir / f"naming-{cid}-{h}.json"
    best = _naming_schema(_json_file(path), len(rows))
    if best is None:
        lines = "\n".join(f"{k + 1}. {_t_text(r)}" for k, r in enumerate(rows))
        kw = dict(topic=topic, members=lines, out_path=path)
        if current is not None:
            kw["current"] = current
        status, best = await run_single_slot(
            ctx, f"Naming {cid}", key=f"blocks-{topic}-naming-{cid}-{h}",
            prompt=_prompt(template, **kw), role="judge", capabilities="files",
            payload=lambda result, p=path, n=len(rows): _naming_schema(_json_file(p), n),
            timeout=_timeout("selection_mapping", len(rows)))
        if status == CANCELLED:
            return ""
        if status == FAILED:
            best = None
    if best 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]
    return rows[best - 1]["norm"]


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 = await _choose_title(ctx, flow, cid, rows, "Blocks-Naming")
        if not winner:  # cancelled
            return
        p["main_norm"] = winner
        w = next(r for r in rows if r["norm"] == winner)
        p["title"], p["description"] = w["title"], w.get("description") or p.get("description", "")
        await db.kanban_set_payload(topic, BOARD, cid, p)
    await db.kanban_advance(topic, BOARD, cid, "naming_check")


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"):
        norms = [r["norm"] for r in rows]
        cur = p.get("main_norm")
        current = norms.index(cur) + 1 if cur in norms else 1
        winner = await _choose_title(ctx, flow, cid, rows, "Blocks-Naming-Check", current=current)
        if not winner:
            return
        w = next(r for r in rows if r["norm"] == winner)
        p["title"], p["description"] = w["title"], 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,
        "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
        fragments: dict[int, int] = {}
        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:
                    fragments[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:
            fragments.pop(nr, None)
        # containment demote + parentless noise (deterministic)
        norms = [(i, allrows[i - 1]["title_norm"]) for i in range(1, n_all + 1)]
        for i in range(1, n_dem + 1):
            if i in fragments or i in honored or 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
        for i in range(1, n_dem + 1):
            if i in fragments or i in honored 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 over still-⚠ survivors (rare-positive recall, majority ≥2).
        # ONE wave over ALL (chunk, judge) slots; a single failed judge is tolerated
        # (panel votes over whatever answered — legacy semantics). Voting afterwards.
        survivors = [i for i in range(1, n_dem + 1)
                     if i not in fragments and i not in honored and _filter_suspect(allrows[i - 1])]
        if survivors:
            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}-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}-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)
                for j in range(1, FILTER_RECHECK_PANEL + 1):
                    raw = _json_file(work_dir / f"filter-recheck-{h}-c{ci}-j{j}.json")
                    v = _filter_schema(raw)
                    if v is None:
                        continue
                    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
                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)
        # 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"})
        atomic_write_json(work_dir / "inventar-filter.json",
                          {"vorher": n_dem, "degradiert": len(journal), "fragments": journal}, 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"], "grouping") 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_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": 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_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:
        status, supplements = await run_single_slot(
            ctx, "Supplement", key=f"blocks-{topic}-supplement",
            prompt=_prompt("Blocks-Supplement", topic=topic,
                           blocks="\n".join(f"- {t}" for t in titles),
                           out_path=path, extra=_extra(flow.state.get("instructions", ""))),
            role="quick", capabilities="full",
            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 = []
    known_norms = set()
    known_keys = set()
    for t in await db.kanban_cards(topic, board=BOARD):
        tt = t["payload"].get("title", "")
        if tt:
            known_norms.add(_norm_title(tt))
            if (k := _canonical_key(tt)):
                known_keys.add(k)
    new = 0
    for t, d in (supplements or []):
        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)
        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:
        _log(topic, f"Supplement: {new} Block-Kandidat(en) → ingest")
        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, "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) -> bool:
    """Run the inventory board (plus board 2 „Artefakte") until quiescence.
    research=False = Continue: drain the existing queue, search nothing new."""
    topic = ctx.topic
    flow = Flow(topic, files["arbeit"])
    flow.state["instructions"] = instructions
    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)
    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)
    stages = chain_stages(stages)
    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))
    try:
        await kanban.run_flow(flow, stages, [_as_producer(p) for p in producers], set_p)
    finally:
        stopper.cancel()
    if ctx.is_cancelled():
        return False
    await _write_final(topic, files)
    return True


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", "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", "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", "grouping", "gap_check", "done"]
_ART_STAGES = ["subblocks", "facts", "levels", "relevance", "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"


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']}"
    if is_active and live_info.get(key):  # live step message wins while the card is worked
        info = live_info[key]
    status = "error" if r["retries"] else ("active" if is_active else "open")
    return {"title": p.get("title") or r["card_id"], "status": status,
            "info": info, "retries": r["retries"]}


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


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)
    return moved


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