"""Streaming kanban dataflow for the inventory phase.

Each column is a worker that pulls cards from its input `stage` (the queue), processes up to
KANBAN_BATCH at a time, and advances them to the next stage. Cards: titles → chains → blocks.

Streaming columns run continuously; barrier columns start only at QUIESCENCE of everything before
them (no active worker + empty queues). Verify columns push failures back (rework). The Chain column
re-clusters live: a chain that gains a member is marked dirty and flows back to chain_verify.

Reused from blocks.py (imported lazily-safe — kanban is only imported after blocks is loaded):
embedding clustering, `_pairs_schema`/`_cliques`, `_canonical`, research prompt + file payload.
"""

import asyncio
import json
import uuid

import database as db
import embedding
import blocks
from config import RESEARCH_GRACE, MAX_CONCURRENT_AGENTS_PER_TOPIC
from pipeline import GenContext, run_single_slot, _prompt, _timeout, _log, OK
from textkit import _norm_title, _title, _parse_selection
from jsonio import read_json_file as _json_file

KANBAN_BATCH = 5          # cards a worker pulls per package (micro-batching)
# How many packages ONE worker keeps in flight at once. A worker no longer blocks on a single
# package — it keeps pulling and dispatching until this many run concurrently, so a busy column
# fills the agent slots (the per-topic semaphore is the real cap; over-dispatch just queues cheaply).
WORKER_INFLIGHT = MAX_CONCURRENT_AGENTS_PER_TOPIC
# Stages whose processor mutates shared cross-card state and MUST run one package at a time.
# Online chain-clustering reads the whole universe + membership; parallel packages would race.
_SERIAL_STAGES = {"chain"}
_ID_COL = {"kanban_titles": "title_norm", "kanban_chains": "chain_id", "kanban_blocks": "block_id"}
CHAIN_CAP = 12            # max members per chain — caps the O(n²) pair-verification blow-up
_POLL = 0.3               # seconds between empty-queue polls

# Stage order. A card's `stage` = the column it waits in (its worker's input).
TITLE_STAGES = ["merge", "chain", "chained"]                 # 'chained' = consumed into a chain
CHAIN_STAGES = ["chain_verify", "naming", "naming_verify", "chain_filter", "filter_verify", "block_assemble"]
BLOCK_STAGES = ["small_blocks", "small_verify", "dependency", "dependency_verify", "main"]
DONE_CHAIN = "done_chain"
DONE_BLOCK = "done_block"
REJECTED = "rejected"     # block dropped by filter_verify (off-topic / noise) — terminal, never mirrored

# Predecessor stages for each barrier (must ALL be quiescent before the barrier worker runs).
_BEFORE_CHAIN_FILTER = ["merge", "chain", "chain_verify", "naming", "naming_verify"]
_BEFORE_BLOCK = _BEFORE_CHAIN_FILTER + ["chain_filter", "filter_verify"]
_BEFORE_MAIN = ["small_blocks", "small_verify", "dependency", "dependency_verify"]


class _Flow:
    """Shared runtime state: active-task counters per stage + a wakeup event. `producers` counts the
    running research agents (initial + any added live via the generate button); research counts as
    done only when ALL producers have finished, so the flow stays awake while extras still search."""
    def __init__(self, topic: str, work_dir):
        self.topic = topic
        self.work_dir = work_dir
        self.active: dict[str, int] = {}
        self.producers = 1          # the initial research agent
        self.research_tag = 0
        self.stop = False
        self.wake = asyncio.Event()
        self.spawn_research = None  # set by run_kanban: () → coroutine that adds one more research agent

    @property
    def research_done(self) -> bool:
        return self.producers <= 0

    def add_producer(self):
        self.producers += 1
        self.wake.set()

    def done_producer(self):
        self.producers -= 1
        self.wake.set()

    def next_tag(self) -> int:
        self.research_tag += 1
        return self.research_tag

    def enter(self, stage: str):
        self.active[stage] = self.active.get(stage, 0) + 1

    def leave(self, stage: str):
        self.active[stage] = max(0, self.active.get(stage, 0) - 1)
        self.wake.set()

    def active_in(self, stages) -> bool:
        return any(self.active.get(s, 0) > 0 for s in stages)

    async def queued_in(self, table: str, stages) -> bool:
        return await db.kanban_count(self.topic, table, list(stages)) > 0


# ── Research producer ────────────────────────────────────────────────────────────
async def _ingest_titles(topic: str, text: str) -> int:
    """Parse a reader file into kanban_titles (stage 'merge'). Exact dupes drop on the PK. → 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)
        parts = [t.strip() for t in record.split(" — ")]
        source = parts[2] if len(parts) >= 3 else ""
        desc = parts[1] if len(parts) >= 2 else ""
        if await db.kanban_add_title(topic, norm, title, source, desc):
            n += 1
    return n


RESEARCH_RUNTIME = 900  # one research agent, one round, ~15 min hard cap — the tail ingests live while it writes
_POLL_RESEARCH = 3      # seconds between live reads of a running research file

# Live registry of running flows, so the "+ research" button can attach another agent to a live run.
_active_flows: dict[str, "_Flow"] = {}


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 _research_once(ctx: GenContext, files: dict, q: dict, folder, instructions: str, tag: str, flow: "_Flow"):
    """ONE agent searches the topic; its titles go into the merge 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 = files["arbeit"]
    caps = "files" if folder else "full"
    p = work_dir / f"research-{tag}.md"
    p.unlink(missing_ok=True)
    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, dupes drop on PK)
        text = "".join(buf)
        if (ft := blocks._file_payload(p)):
            text += "\n" + ft
        return bool(text) and await _ingest_titles(ctx.topic, text)

    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
            if await _drain():
                flow.wake.set()  # new cards → wake the workers

    tail = asyncio.create_task(_tail())
    try:
        await run_single_slot(
            ctx, f"research-{tag}", key=f"blocks-{ctx.topic}-research-{tag}",
            prompt=blocks._build_research_prompt(ctx.topic, p, instructions, q["type"], folder),
            role="quick", capabilities=caps,
            payload=(lambda result, p=p: blocks._file_payload(p)),
            timeout=RESEARCH_RUNTIME, on_line=_on_line,
        )
    finally:
        stop.set()
        await tail
    if await _drain():  # final catch-up
        flow.wake.set()
    _log(ctx.topic, f"Research {tag}: titles → merge queue")


async def _research(ctx: GenContext, files: dict, q: dict, folder, instructions: str, flow: _Flow):
    """The initial research producer (already counted in flow.producers=1)."""
    try:
        await _research_once(ctx, files, q, folder, instructions, "1", flow)
    finally:
        flow.done_producer()


async def _extra_research(ctx: GenContext, files: dict, q: dict, folder, instructions: str, flow: _Flow):
    """One more research agent, added live via the generate button. Keeps the flow awake until done."""
    flow.add_producer()
    try:
        await _research_once(ctx, files, q, folder, instructions, f"x{flow.next_tag()}", flow)
    finally:
        flow.done_producer()


def add_research_agent(topic: str) -> bool:
    """Attach one more research agent to a running flow. → True if a run was live to attach to."""
    flow = _active_flows.get(topic)
    if flow is None or flow.stop or flow.spawn_research is None:
        return False
    asyncio.create_task(flow.spawn_research())
    return True


# ── Generic worker loop ────────────────────────────────────────────────────────────
async def _quiescent(flow: _Flow, stages) -> bool:
    """True iff no worker is active in `stages` AND no card is queued in any of them (all tables).
    The barrier/exit condition — must include QUEUED cards, not just active workers, or a worker
    could exit in a momentary lull while an upstream worker still has work to push down."""
    if not stages:
        return True
    if flow.active_in(stages):
        return False
    for tb in ("kanban_titles", "kanban_chains", "kanban_blocks"):
        if await db.kanban_count(flow.topic, tb, list(stages)):
            return False
    return True


async def _worker(flow: _Flow, table: str, in_stage: str, process, upstream, *, barrier=False, inflight=WORKER_INFLIGHT):
    """Pull cards from `in_stage`, run `process` — keeping up to `inflight` packages running CONCURRENTLY
    so a busy column fills the agent slots instead of doing one package at a time. `upstream` = all stages
    before this one. A barrier worker only pulls when `upstream` is fully quiescent. ANY worker exits only
    when research is done, its own queue is empty, AND `upstream` is quiescent (nothing can still arrive).

    Double-pull safety: each stage has exactly ONE worker, so an in-memory `claimed` set of card-ids (held
    while a package runs) is enough to keep concurrent pulls from grabbing the same cards."""
    topic = flow.topic
    idc = _ID_COL[table]
    claimed: set[str] = set()
    tasks: set[asyncio.Task] = set()

    async def _run(cards):
        ids = [c[idc] for c in cards]
        flow.enter(in_stage)
        try:
            await process(cards)
        except Exception as e:  # one bad package must not kill the worker
            _log(topic, f"worker {in_stage}: {type(e).__name__}: {e}")
        finally:
            flow.leave(in_stage)
            for i in ids:
                claimed.discard(i)
            flow.wake.set()

    try:
        while not flow.stop:
            tasks = {t for t in tasks if not t.done()}
            # Fill the pipeline: pull fresh cards and dispatch until `inflight` packages run.
            if not barrier or await _quiescent(flow, upstream):
                while len(tasks) < inflight:
                    rows = await db.kanban_pull(topic, table, in_stage, KANBAN_BATCH + len(claimed))
                    fresh = [r for r in rows if r[idc] not in claimed][:KANBAN_BATCH]
                    if not fresh:
                        break
                    for r in fresh:
                        claimed.add(r[idc])
                    tasks.add(asyncio.create_task(_run(list(fresh))))
            if tasks:  # busy → wait for a package to finish, then refill
                await asyncio.wait(tasks, timeout=_POLL, return_when=asyncio.FIRST_COMPLETED)
                continue
            # idle: nothing in flight and nothing pulled
            up_quiet = await _quiescent(flow, upstream)
            if (flow.research_done and up_quiet and not flow.active_in([in_stage])
                    and await db.kanban_count(topic, table, in_stage) == 0):
                return  # nothing left and nothing upstream can produce
            await _sleep_wake(flow)
    finally:
        for t in tasks:
            t.cancel()
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)


async def _sleep_wake(flow: _Flow):
    try:
        await asyncio.wait_for(flow.wake.wait(), timeout=_POLL)
    except asyncio.TimeoutError:
        pass
    flow.wake.clear()


# ── Column processors ──────────────────────────────────────────────────────────────
async def _proc_merge(flow: _Flow, cards):
    """Exact dedup happened at ingest (PK). Merge just advances titles to the chain column."""
    for c in cards:
        await db.kanban_advance(flow.topic, "kanban_titles", c["title_norm"], "chain")
    flow.wake.set()


async def _proc_chain(flow: _Flow, cards):
    """Embedding blocking: for each new title, find the most similar existing title (cosine ≥ floor).
    Join its chain (or open a new one), mark the chain dirty → chain_verify. Live-growing clusters."""
    topic = flow.topic
    by_norm = await db.kanban_titles_by_norm(topic)
    # Universe = titles already chained + the new batch (for nearest-neighbour search).
    universe = [nm for nm, r in by_norm.items() if r["stage"] in ("chain", "chained")]
    if len(universe) < 1:
        return
    texts = [f"{by_norm[nm]['title']} — {by_norm[nm]['content']}" if by_norm[nm]["content"] else by_norm[nm]["title"]
             for nm in universe]
    sims = await asyncio.to_thread(embedding.embed_sims, texts) if (
        embedding and await asyncio.to_thread(embedding.available)) else None
    idx = {nm: i for i, nm in enumerate(universe)}
    # existing membership
    member_chain = {}
    for nm in universe:
        cid = await _chain_of(topic, nm)
        if cid:
            member_chain[nm] = cid
    touched = set()
    for c in cards:
        nm = c["title_norm"]
        target = None
        if sims is not None and nm in idx:
            best, bestcos = None, blocks.DEDUP_PAIR_FLOOR
            for other in universe:
                if other == nm or other not in member_chain and other not in idx:
                    continue
                cos = float(sims[idx[nm]][idx[other]]) if other in idx else -1
                if cos >= bestcos and other != nm:
                    best, bestcos = other, cos
            if best is not None:
                target = member_chain.get(best)
        cid = target or f"c-{uuid.uuid4().hex[:12]}"
        members = set(await db.kanban_chain_members(topic, cid))
        if target and len(members) >= CHAIN_CAP:   # neighbour's chain is full → start a fresh chain
            cid = f"c-{uuid.uuid4().hex[:12]}"
            members = set()
        members.add(nm)
        await db.kanban_set_chain_members(topic, cid, sorted(members))
        await db.kanban_upsert_chain(topic, cid, "chain_verify", dirty=1)
        member_chain[nm] = cid
        await db.kanban_advance(topic, "kanban_titles", nm, "chained")
        touched.add(cid)
    flow.wake.set()


async def _chain_of(topic: str, title_norm: str) -> str | None:
    return await db.kanban_member_chain(topic, title_norm)


async def _members_dicts(topic: str, members: list[str]) -> list[dict]:
    by = await db.kanban_titles_by_norm(topic)
    return [by[m] for m in members if m in by]


async def _proc_chain_verify(ctx: GenContext, flow: _Flow, cards):
    """Pairwise-verify the batch's chains IN PARALLEL (one agent per chain). Failures split off."""
    await asyncio.gather(*[_verify_one(ctx, flow, c) for c in cards], return_exceptions=True)
    flow.wake.set()


async def _verify_one(ctx: GenContext, flow: _Flow, c):
    topic = flow.topic
    cid = c["chain_id"]
    members = await db.kanban_chain_members(topic, cid)
    dicts = await _members_dicts(topic, members)
    if len(dicts) <= 1:
        await db.kanban_upsert_chain(topic, cid, "naming", dirty=0)
        return
    # Only the embedding-NEAR candidate pairs (cosine ≥ floor) — NOT all O(n²) pairs. A 12-member
    # chain shrinks from 66 pairs to a handful. Transitivity (connected components) does the rest.
    nm = [d["title_norm"] for d in dicts]
    texts = [f"{d['title']} — {d['content']}" if d["content"] else d["title"] for d in dicts]
    sims = await asyncio.to_thread(embedding.embed_sims, texts) if (
        embedding and await asyncio.to_thread(embedding.available)) else None
    if sims is not None:
        pairs = [(nm[i], nm[j]) for i in range(len(nm)) for j in range(i + 1, len(nm))
                 if float(sims[i][j]) >= blocks.DEDUP_PAIR_FLOOR]
    else:
        pairs = [(a, b) for x, a in enumerate(members) for b in members[x + 1:]]
    keep_edges = await _verify_pairs(ctx, flow.work_dir, topic, cid, dicts, pairs)
    groups = _components(members, keep_edges)  # transitive groups over confirmed near-pairs
    groups.sort(key=len, reverse=True)
    main = groups[0] if groups else members
    await db.kanban_set_chain_members(topic, cid, sorted(main))
    await db.kanban_upsert_chain(topic, cid, "naming", dirty=0)
    for g in groups[1:]:  # the rest split into fresh chains
        ncid = f"c-{uuid.uuid4().hex[:12]}"
        await db.kanban_set_chain_members(topic, ncid, sorted(g))
        await db.kanban_upsert_chain(topic, ncid, "naming", dirty=0)


def _components(members: list[str], edges) -> list[list[str]]:
    """Connected components (union-find) over confirmed pairs. Members without an edge stay alone."""
    parent = {m: m for m in members}

    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    for a, b in edges:
        if a in parent and b in parent:
            parent[find(a)] = find(b)
    comp: dict[str, list[str]] = {}
    for m in members:
        comp.setdefault(find(m), []).append(m)
    return list(comp.values())


async def _verify_pairs(ctx, work_dir, topic, cid, dicts, pairs):
    """Judge the candidate pairs in DEDUP_PAIRS_CHUNK packages, all packages IN PARALLEL → confirmed edges."""
    by = {d["title_norm"]: d for d in dicts}
    chunks = [pairs[k:k + blocks.DEDUP_PAIRS_CHUNK] for k in range(0, len(pairs), blocks.DEDUP_PAIRS_CHUNK)]

    async def _chunk(ci, chunk):
        path = work_dir / f"verify-{cid}-{ci}.json"
        lines = "\n\n".join(
            f"{j + 1}.\nA: {by[a]['title']} — {by[a]['content']}\nB: {by[b]['title']} — {by[b]['content']}"
            for j, (a, b) in enumerate(chunk))
        await run_single_slot(
            ctx, f"Chain verify {cid}", key=f"blocks-{topic}-verify-{cid}-{ci}",
            prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=path),
            role="judge", capabilities="files",
            payload=lambda result, p=path: blocks._pairs_schema(_json_file(p)),
            timeout=_timeout("selection_mapping", len(chunk)))
        verdict = blocks._pairs_schema(_json_file(path)) or {}
        return [(a, b) for j, (a, b) in enumerate(chunk) if verdict.get(j + 1)]

    results = await asyncio.gather(*[_chunk(ci, ch) for ci, ch in enumerate(chunks)], return_exceptions=True)
    return [e for r in results if isinstance(r, list) for e in r]


async def _proc_naming(ctx: GenContext, flow: _Flow, cards):
    """Pick the best member title per chain — batch runs IN PARALLEL (one agent per chain)."""
    await asyncio.gather(*[_name_one(ctx, flow, c) for c in cards], return_exceptions=True)
    flow.wake.set()


async def _name_one(ctx: GenContext, flow: _Flow, c):
    topic = flow.topic
    cid = c["chain_id"]
    members = await db.kanban_chain_members(topic, cid)
    dicts = await _members_dicts(topic, members)
    if len(dicts) <= 1:
        await db.kanban_upsert_chain(topic, cid, "naming_verify",
                                     main_title_norm=(members[0] if members else None), dirty=0)
        return
    winner = await _choose_title(ctx, flow.work_dir, topic, cid, members, dicts, "Blocks-Naming")
    await db.kanban_upsert_chain(topic, cid, "naming_verify", main_title_norm=winner, dirty=0)


async def _proc_naming_verify(ctx: GenContext, flow: _Flow, cards):
    """Second judge checks each title — batch runs IN PARALLEL."""
    await asyncio.gather(*[_namecheck_one(ctx, flow, c) for c in cards], return_exceptions=True)
    flow.wake.set()


async def _namecheck_one(ctx: GenContext, flow: _Flow, c):
    topic = flow.topic
    cid = c["chain_id"]
    members = await db.kanban_chain_members(topic, cid)
    dicts = await _members_dicts(topic, members)
    winner = c.get("main_title_norm") or (members[0] if members else None)
    if len(dicts) > 1:
        winner = await _choose_title(ctx, flow.work_dir, topic, cid, members, dicts, "Blocks-Naming-Check",
                                     current=(members.index(winner) + 1 if winner in members else 1))
    await db.kanban_upsert_chain(topic, cid, "chain_filter", main_title_norm=winner, dirty=0)


async def _choose_title(ctx, work_dir, topic, cid, members, dicts, template, current=None):
    by = {d["title_norm"]: d for d in dicts}
    path = work_dir / f"naming-{cid}.json"
    lines = "\n".join(f"{k + 1}. {by[m]['title']} — {by[m]['content']}" for k, m in enumerate(members) if m in by)
    kw = dict(topic=topic, members=lines, out_path=path)
    if current is not None:
        kw["current"] = current
    await run_single_slot(
        ctx, f"Naming {cid}", key=f"blocks-{topic}-naming-{cid}",
        prompt=_prompt(template, **kw), role="judge", capabilities="files",
        payload=lambda result, p=path: blocks._naming_schema(_json_file(p), len(members)),
        timeout=_timeout("selection_mapping", len(members)))
    best = blocks._naming_schema(_json_file(path), len(members))
    if best is None:
        rep = blocks._canonical(dicts, list(range(len(dicts))), set())
        w = _norm_title(rep["title"])
        return w if w in members else members[0]
    return members[best - 1]


async def _proc_chain_filter(flow: _Flow, cards):
    """BARRIER. Reduce each chain to its winner → upsert a block (chain_id stable). → filter_verify."""
    topic = flow.topic
    by = await db.kanban_titles_by_norm(topic)
    for c in cards:
        cid = c["chain_id"]
        members = await db.kanban_chain_members(topic, cid)
        winner = c.get("main_title_norm") if c.get("main_title_norm") in members else (members[0] if members else None)
        if not winner or winner not in by:
            await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0)
            continue
        w = by[winner]
        await db.kanban_upsert_block(topic, f"b-{cid}", cid, w["title"], w["source"], w["content"], stage="filter_verify_b")
        await db.kanban_upsert_chain(topic, cid, "filter_verify", dirty=0)
    flow.wake.set()


async def _proc_filter_verify(ctx: GenContext, flow: _Flow, cards):
    """An agent confirms each reduced block is a valid, on-topic, self-contained concept. Off-topic /
    noise / empty blocks are dropped (→ REJECTED, chain done). IN PARALLEL (one agent per block)."""
    await asyncio.gather(*[_filtercheck_one(ctx, flow, c) for c in cards], return_exceptions=True)
    flow.wake.set()


async def _filtercheck_one(ctx: GenContext, flow: _Flow, c):
    topic = flow.topic
    cid = c["chain_id"]
    bid = f"b-{cid}"
    by = await db.kanban_titles_by_norm(topic)
    members = await db.kanban_chain_members(topic, cid)
    winner = c.get("main_title_norm") if c.get("main_title_norm") in by else (members[0] if members else None)

    async def _drop():
        await db.kanban_advance(topic, "kanban_blocks", bid, REJECTED)
        await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0)

    async def _pass():
        await db.kanban_advance(topic, "kanban_blocks", bid, "block_assemble_b")
        await db.kanban_upsert_chain(topic, cid, "block_assemble", dirty=0)

    if not winner or winner not in by:  # nothing to verify → drop the empty chain
        await _drop()
        return
    w = by[winner]
    path = flow.work_dir / f"filtercheck-{cid}.json"
    await run_single_slot(
        ctx, f"Filter verify {cid}", key=f"blocks-{topic}-verify-filter-{cid}",
        prompt=_prompt("Blocks-Filter-Check", topic=topic, title=w["title"], content=w["content"], out_path=path),
        role="judge", capabilities="files",
        payload=lambda result, p=path: _keep_schema(_json_file(p)),
        timeout=_timeout("selection_mapping", 1))
    keep = _keep_schema(_json_file(path))
    await (_drop() if keep is False else _pass())  # None (parse fail) → keep, conservative


async def _proc_block(flow: _Flow, cards):
    """BARRIER. Assemble the final block row → small_blocks. (chain card consumed → done.)"""
    topic = flow.topic
    for c in cards:
        cid = c["chain_id"]
        await db.kanban_advance(topic, "kanban_blocks", f"b-{cid}", "small_blocks")
        await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0)
    flow.wake.set()


def _small_schema(data, count):
    """{"small": {"1": true, ...}} → {block_index: bool} · else None."""
    if not isinstance(data, dict) or not isinstance(data.get("small"), dict):
        return None
    out = {}
    for k, v in data["small"].items():
        try:
            n = int(k)
        except (ValueError, TypeError):
            continue
        if 1 <= n <= count:
            out[n] = str(v).strip().casefold() in ("true", "ja", "yes", "1")
    return out or None


def _keep_schema(data):
    """{"keep": true/false} → bool · None when absent/unparseable (caller keeps on None, conservative)."""
    if not isinstance(data, dict) or "keep" not in data:
        return None
    return str(data["keep"]).strip().casefold() in ("true", "ja", "yes", "1")


def _dep_schema(data, count):
    """{"parent": N} → 0..count (0 = standalone) · else None."""
    if not isinstance(data, dict):
        return None
    try:
        n = int(data.get("parent"))
    except (ValueError, TypeError):
        return None
    return n if 0 <= n <= count else None


async def _proc_small(ctx: GenContext, flow: _Flow, cards):
    """Judge marks fragment-like blocks (batch). → small_verify."""
    topic = flow.topic
    path = flow.work_dir / f"small-{cards[0]['block_id']}.json"
    lines = "\n".join(f"{i + 1}. {c['title']} — {c['content']}" for i, c in enumerate(cards))
    await run_single_slot(
        ctx, "Small blocks", key=f"blocks-{topic}-small-{cards[0]['block_id']}",
        prompt=_prompt("Blocks-Small", topic=topic, blocks=lines, out_path=path),
        role="judge", capabilities="files",
        payload=lambda result, p=path: _small_schema(_json_file(p), len(cards)),
        timeout=_timeout("selection_mapping", len(cards)))
    verdict = _small_schema(_json_file(path), len(cards)) or {}
    for i, c in enumerate(cards):
        is_small = 1 if verdict.get(i + 1) else 0
        await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"],
                                     stage="small_verify", is_small=is_small, parent_block_id=c.get("parent_block_id"))
    flow.wake.set()


async def _proc_small_verify(ctx: GenContext, flow: _Flow, cards):
    """Second judge re-checks the small flag (consensus): a block stays `small` only if it was marked
    small AND this judge also calls it a fragment. Disagreement → keep as a main block (conservative).
    → dependency."""
    topic = flow.topic
    path = flow.work_dir / f"smallcheck-{cards[0]['block_id']}.json"
    lines = "\n".join(f"{i + 1}. {c['title']} — {c['content']}" for i, c in enumerate(cards))
    await run_single_slot(
        ctx, "Small verify", key=f"blocks-{topic}-verify-small-{cards[0]['block_id']}",
        prompt=_prompt("Blocks-Small", topic=topic, blocks=lines, out_path=path),
        role="judge", capabilities="files",
        payload=lambda result, p=path: _small_schema(_json_file(p), len(cards)),
        timeout=_timeout("selection_mapping", len(cards)))
    verdict = _small_schema(_json_file(path), len(cards)) or {}
    for i, c in enumerate(cards):
        is_small = 1 if (c["is_small"] and verdict.get(i + 1)) else 0
        await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"],
                                     stage="dependency", is_small=is_small, parent_block_id=c.get("parent_block_id"))
    flow.wake.set()


async def _proc_dependency(ctx: GenContext, flow: _Flow, cards):
    """For each SMALL block, a judge picks its parent from the full list — batch runs IN PARALLEL."""
    topic = flow.topic
    parents = [b for b in await db.kanban_all_blocks(topic) if not b["is_small"] and b["stage"] != REJECTED]
    plist = "\n".join(f"{i + 1}. {b['title']}" for i, b in enumerate(parents))
    await asyncio.gather(*[_dep_one(ctx, flow, c, parents, plist) for c in cards], return_exceptions=True)
    flow.wake.set()


async def _dep_one(ctx: GenContext, flow: _Flow, c, parents, plist):
    topic = flow.topic
    if not c["is_small"] or not parents:
        await db.kanban_advance(topic, "kanban_blocks", c["block_id"], "dependency_verify")
        return
    path = flow.work_dir / f"dep-{c['block_id']}.json"
    await run_single_slot(
        ctx, "Dependency", key=f"blocks-{topic}-dep-{c['block_id']}",
        prompt=_prompt("Blocks-Dependency", topic=topic,
                       small=f"{c['title']} — {c['content']}", parents=plist, out_path=path),
        role="judge", capabilities="files",
        payload=lambda result, p=path: _dep_schema(_json_file(p), len(parents)),
        timeout=_timeout("selection_mapping", len(parents)))
    pick = _dep_schema(_json_file(path), len(parents))
    parent_id = parents[pick - 1]["block_id"] if pick else None
    await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"],
                                 stage="dependency_verify", is_small=c["is_small"], parent_block_id=parent_id)


async def _proc_dependency_verify(flow: _Flow, cards):
    """A small block without a parent is demarked → becomes a main block. → main."""
    topic = flow.topic
    for c in cards:
        is_small = c["is_small"]
        if is_small and not c.get("parent_block_id"):
            is_small = 0
        await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"],
                                     stage="main", is_small=is_small, parent_block_id=c.get("parent_block_id"))
    flow.wake.set()


async def _proc_main(flow: _Flow, cards):
    """BARRIER. Finalize non-small blocks → mirror into the legacy `blocks` table as consensus."""
    topic = flow.topic
    for c in cards:
        if not c["is_small"]:
            norm = _norm_title(c["title"])
            await db.upsert_block(topic, norm, c["title"], c["content"], [c["source"]] if c["source"] else [])
            await db.set_block_status(topic, norm, "consensus")
        await db.kanban_advance(topic, "kanban_blocks", c["block_id"], DONE_BLOCK)
    flow.wake.set()


# ── Orchestration ──────────────────────────────────────────────────────────────────
async def run_kanban(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str,
                     research: bool = True) -> bool:
    """Run the streaming inventory. Returns True when the whole flow reaches quiescence at 'main'.

    research=False ("Continue"): process the EXISTING queue without searching new titles. No initial
    research producer, producers=0 → research_done is true at once; workers drain the queue and exit.
    The +Research button can still attach an agent later via flow.spawn_research."""
    topic = ctx.topic
    flow = _Flow(topic, files["arbeit"])
    flow.spawn_research = lambda: _extra_research(ctx, files, q, folder, instructions, flow)
    if not research:
        flow.producers = 0  # continue the existing queue, search no new titles
    _active_flows[topic] = flow
    set_p("Kanban inventory…")

    ORDER = ["merge", "chain", "chain_verify", "naming", "naming_verify", "chain_filter",
             "filter_verify", "block_assemble", "small_blocks", "small_verify",
             "dependency", "dependency_verify", "main"]
    up = {s: ORDER[:i] for i, s in enumerate(ORDER)}   # upstream = all stages before this one
    barriers = {"chain_filter", "block_assemble", "main"}
    specs = [
        ("kanban_titles", "merge", lambda cs: _proc_merge(flow, cs)),
        ("kanban_titles", "chain", lambda cs: _proc_chain(flow, cs)),
        ("kanban_chains", "chain_verify", lambda cs: _proc_chain_verify(ctx, flow, cs)),
        ("kanban_chains", "naming", lambda cs: _proc_naming(ctx, flow, cs)),
        ("kanban_chains", "naming_verify", lambda cs: _proc_naming_verify(ctx, flow, cs)),
        ("kanban_chains", "chain_filter", lambda cs: _proc_chain_filter(flow, cs)),
        ("kanban_chains", "filter_verify", lambda cs: _proc_filter_verify(ctx, flow, cs)),
        ("kanban_chains", "block_assemble", lambda cs: _proc_block(flow, cs)),
        ("kanban_blocks", "small_blocks", lambda cs: _proc_small(ctx, flow, cs)),
        ("kanban_blocks", "small_verify", lambda cs: _proc_small_verify(ctx, flow, cs)),
        ("kanban_blocks", "dependency", lambda cs: _proc_dependency(ctx, flow, cs)),
        ("kanban_blocks", "dependency_verify", lambda cs: _proc_dependency_verify(flow, cs)),
        ("kanban_blocks", "main", lambda cs: _proc_main(flow, cs)),
    ]
    workers = [_research(ctx, files, q, folder, instructions, flow)] if research else []
    for table, stage, proc in specs:
        workers.append(_worker(flow, table, stage, proc, up[stage], barrier=(stage in barriers),
                               inflight=(1 if stage in _SERIAL_STAGES else WORKER_INFLIGHT)))

    progress = asyncio.create_task(_progress(flow, set_p))
    try:
        await asyncio.gather(*workers, return_exceptions=True)
    finally:
        flow.stop = True
        progress.cancel()
        _active_flows.pop(topic, None)
    if ctx.is_cancelled():
        return False
    n = await db.kanban_count(topic, "kanban_blocks", DONE_BLOCK)
    _log(topic, f"Kanban: done — {n} blocks finalized")
    return True


async def _progress(flow: _Flow, set_p):
    while not flow.stop:
        try:
            counts = await db.kanban_stage_counts(flow.topic)
            total = sum(counts.values())
            set_p(f"Kanban: {total} cards in flow")
        except Exception:
            pass
        await asyncio.sleep(1.0)