"""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, FRAGMENT_MIN_COS, GROUP_MIN_COS_FLOOR, GROUP_RECONCILE_FLOOR, ) from fsutil import atomic_write_json, atomic_write_text from jsonio import read_json_file as _json_file from pipeline import (CANCELLED, FAILED, OK, GenContext, _extra, _log, _prompt, _runde_schema, _timeout, 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 proposals: dict[int, int] = {} # judge demotes are PROPOSALS — panel/containment confirm drops: set[int] = set() for ci, numbers in enumerate(chunks): raw = _json_file(work_dir / f"filter-{h}-c{ci}.json") verdict = _filter_schema(raw) or {} nset = set(numbers) for nr, parent in verdict.items(): if 1 <= parent <= n_all and nr in nset: proposals[nr] = parent for x in (raw.get("drop", []) if isinstance(raw, dict) else []): try: dnr = int(x) except (ValueError, TypeError): continue if dnr in nset: drops.add(dnr) # hard-drop double gate honored = {nr for nr in drops if _is_artifact(allrows[nr - 1]["title"])} for nr in honored: proposals.pop(nr, None) # containment demote (deterministic — auto-confirms proposals whose parent name is # literally contained in the title, and keeps the legacy ⚠-only pass) + parentless noise norms = [(i, allrows[i - 1]["title_norm"]) for i in range(1, n_all + 1)] fragments: dict[int, int] = {} contained: set[int] = set() for i in range(1, n_dem + 1): if i in honored or (i not in proposals and not _filter_suspect(allrows[i - 1])): continue parent = _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i]) if parent is not None and parent != i and parent not in honored: fragments[i] = parent contained.add(i) proposals.pop(i, None) for i in range(1, n_dem + 1): if i in fragments or i in honored or i in proposals or not _is_parentless_noise(allrows[i - 1]["title"]): continue if _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i]) is None: honored.add(i) # recheck panel = second opinion: unconfirmed judge proposals (WITHOUT the suggested # parent — no anchoring) plus still-⚠ survivors. Majority ≥2 demotes/drops; a proposal # the panel does not confirm survives. The panel is load-bearing now: fewer than 2 # valid judge files per chunk is an error (backoff), not a silent keep. survivors = [i for i in range(1, n_dem + 1) if i not in fragments and i not in honored and (i in proposals or _filter_suspect(allrows[i - 1]))] overruled: list[int] = [] if survivors: ph = _h(",".join(map(str, survivors))) # panel-input hash: stale pre-change files never match rchunks = [survivors[k:k + FILTER_CHUNK] for k in range(0, len(survivors), FILTER_CHUNK)] async def _recheck_judge(ci, nums, j): path = work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json" if _filter_schema(_json_file(path)) is not None: return # resume await run_single_slot( ctx, f"Filter-Recheck {ci}/{j}", key=f"blocks-{topic}-filter-recheck-{h}-{ph}-c{ci}-j{j}", prompt=_prompt("Blocks-Filter-Recheck", topic=topic, survivors="\n".join(_fline(i) for i in nums), list=full_list, out_path=path), role="judge", capabilities="files", payload=lambda result, p=path: _filter_schema(_json_file(p)), timeout=_timeout("selection_mapping", len(nums))) await asyncio.gather(*[ _recheck_judge(ci, nums, j) for ci, nums in enumerate(rchunks) for j in range(1, FILTER_RECHECK_PANEL + 1)], return_exceptions=True) if ctx.is_cancelled(): return for ci, nums in enumerate(rchunks): dem: dict[int, list[int]] = {} drp: dict[int, int] = {} nset = set(nums) valid = 0 for j in range(1, FILTER_RECHECK_PANEL + 1): raw = _json_file(work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json") v = _filter_schema(raw) if v is None: continue valid += 1 for nr, parent in v.items(): if nr in nset and 1 <= parent <= n_all and nr != parent: dem.setdefault(nr, []).append(parent) for x in (raw.get("drop", []) if isinstance(raw, dict) else []): try: dnr = int(x) except (ValueError, TypeError): continue if dnr in nset: drp[dnr] = drp.get(dnr, 0) + 1 if valid < 2: raise RuntimeError(f"Filter-Recheck chunk {ci}: nur {valid} Judge(s) mit Ergebnis") for nr in nums: if nr in fragments or nr in honored: continue if drp.get(nr, 0) >= 2 and (_is_artifact(allrows[nr - 1]["title"]) or _is_parentless_noise(allrows[nr - 1]["title"])): honored.add(nr) elif len(dem.get(nr, [])) >= 2: fragments[nr] = max(set(dem[nr]), key=dem[nr].count) elif nr in proposals: overruled.append(nr) # embedding backstop: veto confirmed non-containment demotes whose direct title pair is # literally structureless (see FRAGMENT_MIN_COS) — applied BEFORE _root resolution. floor_veto: list[int] = [] if (cand := [nr for nr in fragments if nr not in contained]): va = await _vec_rows(flow, [r["title"] for r in allrows]) if va is not None: for nr in cand: if float(va[nr - 1] @ va[fragments[nr] - 1]) < FRAGMENT_MIN_COS: fragments.pop(nr) floor_veto.append(nr) # statement-gate rescue (final override) def _protected(nr): return _is_named_statement(allrows[nr - 1]["title"], allrows[nr - 1]["description"]) for nr in [nr for nr in list(fragments) if _protected(nr)]: fragments.pop(nr, None) honored -= {nr for nr in honored if _protected(nr)} # transitive resolution + moves journal = [] for nr in sorted(fragments): root, cyclic = _root(nr, fragments) if cyclic or not (1 <= root <= n_all) or nr > n_dem: continue r = allrows[nr - 1] parent_norm = None if root in honored else allrows[root - 1]["title_norm"] r["payload"].update(reason="fragment" if parent_norm else "drop-collateral", parent_norm=parent_norm) await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"]) moves.append((r["card_id"], "rejected")) journal.append({"fragment": r["title"], "eltern": allrows[root - 1]["title"] if parent_norm else None}) for nr in sorted(honored): if nr > n_dem: continue r = allrows[nr - 1] r["payload"].update(reason="drop", parent_norm=None) await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"]) moves.append((r["card_id"], "rejected")) journal.append({"fragment": r["title"], "eltern": None, "grund": "drop"}) # one journal file per pass (h) — the supplement feedback pass must not overwrite # the main pass's journal (it is the evaluation instrument). atomic_write_json(work_dir / f"inventar-filter-{h}.json", {"vorher": n_dem, "degradiert": len(journal), "fragments": journal, "ueberstimmt": [allrows[nr - 1]["title"] for nr in overruled], "floor_veto": [allrows[nr - 1]["title"] for nr in floor_veto]}, indent=1) _log(topic, f"Fragment-Filter: {n_dem} → {n_dem - len(journal)} (−{len(journal)})") demoted = {cid for cid, _ in moves} moves += [(r["card_id"], "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 = [] # Dead lineage: blocks demoted by the fragment filter (and their cluster + title cards) # must NOT dedup a supplement proposal — their content is gone. A hit on a dead title # REOPENS the lineage instead: the title card rejoins its cluster (live re-cluster) and # the respawned block gets a fresh fragment_filter pass. failed-quorum/pre-reject stay # in the dedup: those were rejected as non-blocks, not lost as content. dead_reasons = {"fragment", "drop-collateral", "drop"} cards = await db.kanban_cards(topic, board=BOARD) dead_clusters = {c["payload"].get("cluster") for c in cards if c["kind"] == "block" and c["stage"] == "rejected" and c["payload"].get("reason") in dead_reasons} dead_clusters.discard(None) membership = await db.kanban_membership(topic) dead_titles = {nm for nm, cid in membership.items() if cid in dead_clusters} def _is_dead(c) -> bool: if c["kind"] == "title": return c["card_id"] in dead_titles if c["kind"] == "cluster": return c["card_id"] in dead_clusters return c["stage"] == "rejected" and c["payload"].get("reason") in dead_reasons known_norms, known_keys = set(), set() dead_by_norm: dict[str, str] = {} # title norm/key → requeue-able title card_id dead_by_key: dict[str, str] = {} for c in cards: tt = c["payload"].get("title", "") if not tt: continue if _is_dead(c): if c["kind"] == "title": dead_by_norm.setdefault(c["card_id"], c["card_id"]) if (k := _canonical_key(tt)): dead_by_key.setdefault(k, c["card_id"]) continue known_norms.add(_norm_title(tt)) if (k := _canonical_key(tt)): known_keys.add(k) new = reopened = 0 for t, d in (supplements or []): norm = _norm_title(t) key = _canonical_key(t) if not norm or norm in known_norms or (key and key in known_keys): continue known_norms.add(norm) dead_id = dead_by_norm.get(norm) or (dead_by_key.get(key) if key else None) if dead_id: card = await db.kanban_get_card(topic, BOARD, dead_id) if card: card["payload"]["supplement"] = True await db.kanban_set_payload(topic, BOARD, dead_id, card["payload"]) await db.kanban_advance(topic, BOARD, dead_id, "cluster") reopened += 1 continue desc = f"{d} [Supplement]".strip() async with _ingest_lock: await db.kanban_add_title(topic, BOARD, norm, t, desc, "supplement", "supplement") card = await db.kanban_get_card(topic, BOARD, norm) if card: card["payload"]["supplement"] = True await db.kanban_set_payload(topic, BOARD, norm, card["payload"]) new += 1 if new or reopened: _log(topic, f"Supplement: {new} Block-Kandidat(en) → ingest, {reopened} wiedereröffnet") flow.wake.set() async def _proc_done(ctx: GenContext, flow: Flow, cards): """Mirror finished blocks into the legacy `blocks` table (status consensus) — the interface everything downstream (guide, exam, frontend) already reads. Unique-title collisions get a numeric suffix; a re-run under a new name discards the old mirror.""" topic = flow.topic mirrored: dict[str, str] = flow.state.setdefault("mirrored", {}) moves = [] for c in cards: p = c["payload"] base = p.get("title", "") if not base: moves.append((c["card_id"], "done_block")) continue title, norm, i = base, _norm_title(base), 2 while mirrored.get(norm) not in (None, c["card_id"]): title = f"{base} ({i})" norm = _norm_title(title) i += 1 old = p.get("mirrored_norm") if old and old != norm: await db.set_block_status(topic, old, "discarded") await db.upsert_block(topic, norm, title, p.get("description", ""), p.get("sources") or []) await db.set_block_status(topic, norm, "consensus", title=title, description=p.get("description", "")) mirrored[norm] = c["card_id"] p.update(mirrored_norm=norm, title=title) await db.kanban_set_payload(topic, BOARD, c["card_id"], p) if (spawn := flow.state.get("spawn_artefact")): await spawn(c["card_id"], p) # board 2 card per finished block moves.append((c["card_id"], "done_block")) await db.kanban_advance_many(topic, BOARD, moves) flow.wake.set() # ── Orchestration ────────────────────────────────────────────────────────────────── def inventory_stages(ctx: GenContext, flow: Flow) -> list[Stage]: research_done = lambda: flow.research_done # noqa: E731 return [ Stage(BOARD, "ingest", lambda cs: _proc_ingest(flow, cs)), Stage(BOARD, "cluster", lambda cs: _proc_cluster(ctx, flow, cs), serial=True), Stage(BOARD, "pair_check", lambda cs: _proc_pair_check(ctx, flow, cs)), # gate (no barrier): hold clusters until ALL research producers are done — a title's # second reader may arrive minutes later, and the ≥2 vote must count it. Stage(BOARD, "consensus_gate", lambda cs: _proc_consensus_gate(ctx, flow, cs), gate=research_done), Stage(BOARD, "clarify", lambda cs: _proc_clarify(ctx, flow, cs)), Stage(BOARD, "naming", lambda cs: _proc_naming(ctx, flow, cs)), Stage(BOARD, "naming_check", lambda cs: _proc_naming_check(ctx, flow, cs)), Stage(BOARD, "fragment_filter", lambda cs: _proc_fragment_filter(ctx, flow, cs), barrier=True, drain=True, gate=research_done), Stage(BOARD, "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) if artefacts: # Outline needs every block's TITLE + FACTS, nothing later: cut the post-facts # artefact stages from its barrier so it runs parallel to levels…finalize of the # slowest block (makespan tail). Inventory stages all stay — no late blocks. outline = next(s for s in stages if s.stage == "outline") outline.upstream = [u for u in outline.upstream if u not in ("levels", "relevance", "question_pattern", "artefacts", "finalize")] producers = _build_producers(ctx, flow, q, folder, instructions) if research else [] async def _as_producer(coro): try: await coro except Exception: log.exception("[%s] research producer failed", topic) finally: flow.done_producer() for _ in producers: flow.add_producer() flow.spawn_research = lambda: _research_once( ctx, flow, q, folder, instructions, f"x{flow.next_tag()}") # cancel hook: blocks.cancel_blocks flips is_cancelled; stop the flow with it stopper = asyncio.create_task(_stop_on_cancel(ctx, flow)) 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" # Board-2 stage → its fine-step group in blocks.PHASEN (drives the per-card stepper). _STAGE_PHASE = {"subblocks": "Subblocks", "facts": "Facts", "levels": "Levels", "relevance": "Relevance", "question_pattern": "Questions", "artefacts": "Artefacts"} _PHASE_STEPS = {name: steps for name, steps in blocks.PHASEN} def _card_view(r: dict, active: set[str], live_info: dict) -> dict: p = r["payload"] key = f"{r['board']}:{r['card_id']}" is_active = key in active info = r.get("last_error") or p.get("reason") or "" if p.get("merged_into"): info = f"→ {p['merged_into']}" elif p.get("parent_norm"): info = f"Fragment von: {p['parent_norm']}" out = {"title": p.get("title") or r["card_id"], "retries": r["retries"], "card_id": r["card_id"], "kind": r.get("kind", ""), "board": r.get("board", ""), "status": "error" if r["retries"] else ("active" if is_active else "open")} live = live_info.get(key) if is_active and live: # live step message wins while the card is worked info = live["msg"] if isinstance(live, dict) else live # dict since the stepper, str before step = live.get("step") if isinstance(live, dict) else "" steps = _PHASE_STEPS.get(_STAGE_PHASE.get(r.get("stage", ""), ""), ()) if step in steps: # phase stepper: which fine step of the card's stage runs (1-based) out.update(step_i=steps.index(step) + 1, step_n=len(steps), steps=list(steps)) out["info"] = info return out async def board_snapshot(topic: str, limit: int = 20) -> dict: """Live board: per column count + the newest cards (title, status, info).""" counts = await db.kanban_stage_counts(topic) flow = kanban.active_flows.get(topic) active = flow.active_cards if flow else set() live_info = flow.state.get("card_info", {}) if flow else {} columns = [] for board, stage, label, _kind in COLUMNS: total = counts.get(board, {}).get(stage, 0) cards = ([_card_view(r, active, live_info) for r in await db.kanban_stage_cards(topic, board, stage, limit)] if total else []) columns.append({"board": board, "key": stage, "label": label, "total": total, "cards": cards}) dead = [{"card_id": r["card_id"], "board": r["board"], "kind": r["kind"], "title": r["payload"].get("title") or r["card_id"], "error": r.get("last_error") or ""} for r in await db.kanban_dead(topic)] return {"columns": columns, "dead": dead, "done": counts.get("inventory", {}).get("done_block", 0)} 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) await db.add_event(topic, "reset", key=f"{board}:from-{stage}", status=str(moved)) return moved async def restart_artefact_card(topic: str, card_id: str) -> bool: """Restart ONE artefacts card from `subblocks` — wipes only ITS derived DB rows (per-block work-dir slots overwrite themselves; finalize re-upserts later). Only call while nothing is generating (the route guards). → False if unknown.""" card = await db.kanban_get_card(topic, "artefacts", card_id) if card is None or card["kind"] != "ablock": return False await db.delete_subblocks(topic, card_id) await db.delete_question_pattern(topic, card_id) await db.delete_sub_artefakte(topic, card_id) p = {k: v for k, v in card["payload"].items() if k in ("title", "description")} await db.kanban_set_payload(topic, "artefacts", card_id, p) await db.kanban_advance(topic, "artefacts", card_id, "subblocks") await db.add_event(topic, "reset", key=f"artefacts:{card_id}", status="card-restart") return True async def requeue_dead(topic: str) -> int: """Dead-letter → restart stage by card kind (fresh retries). → requeued count.""" n = 0 for r in await db.kanban_dead(topic): stage = _DEAD_RESTART.get(r["kind"]) if stage: await db.kanban_advance(topic, r["board"], r["card_id"], stage) n += 1 return n def add_research_agent(topic: str) -> bool: """Attach one more research agent to a live flow. Counted SYNCHRONOUSLY before the task (quiescence race). → True if a run was live to attach to.""" flow = kanban.active_flows.get(topic) if flow is None or flow.stop or flow.spawn_research is None: return False flow.add_producer() async def _run(): try: await flow.spawn_research() except Exception: log.exception("[%s] extra research failed", topic) finally: flow.done_producer() asyncio.create_task(_run()) return True