1925 lines
95 KiB
Python
1925 lines
95 KiB
Python
"""Board 1 „Inventar": streaming kanban stages for the block inventory.
|
||
|
||
Research producers stream candidate titles into the board; the columns pull, filter,
|
||
merge and reject them until an optimized block inventory remains. The mature filter
|
||
logic (relation guard, complete-link cliques, degrade pass, umbrella grouping, gates)
|
||
is reused from blocks.py — this module only re-orchestrates it as streaming stages.
|
||
|
||
Stages (cards):
|
||
ingest title exact dedup folded at DB-add (reader union)
|
||
cluster title SERIAL — online embedding nearest-neighbour clustering
|
||
pair_check cluster judge per candidate pair (relation guard, cliques) → splits
|
||
consensus_gate cluster code: reader union ≥2 (or supplement) → naming, else clarify
|
||
clarify cluster 3-judge panel, unanimity for single-reader finds
|
||
naming cluster judge picks the canonical member title
|
||
naming_check cluster second judge verifies → spawns the block card
|
||
fragment_filter block BARRIER/drain: global re-merge + degrade pass (full list)
|
||
dedup block BARRIER/drain: global judge-verified pair dedup (incl. context)
|
||
grouping block BARRIER/drain: umbrella grouping (type gate, reconcile)
|
||
gap_check block BARRIER/drain: one supplement round (web) → feeds ingest
|
||
done block mirror into the legacy `blocks` table → done_block
|
||
|
||
Terminal: clustered (titles), done_cluster, grouped, rejected (with journal), done_block.
|
||
"""
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import math
|
||
import re
|
||
import unicodedata
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
import database as db
|
||
import embedding
|
||
import kanban
|
||
from kanban import Flow, Stage, chain_stages
|
||
import blocks
|
||
from blocks import (
|
||
DEDUP_GLOBAL_FLOOR, DEDUP_PAIR_FLOOR, DEDUP_PAIRS_CHUNK, DEDUP_TITLE_AUTO, FILTER_CHUNK,
|
||
FILTER_RECHECK_PANEL, CONSOLIDATION_PANEL, RESEARCH_BATCH, RESEARCH_READERS,
|
||
RESEARCH_THEMA_AGENTS, _FILTER_NOTATION, _GROUP_STANDALONE,
|
||
_build_research_prompt, _canonical, _canonical_key, _chunk_nums, _cliques,
|
||
_completion_schema, _containment_parent, _crawl_index, _file_payload,
|
||
_filter_schema, _filter_suspect, _is_artifact, _is_named_statement,
|
||
_is_parentless_noise, _is_reference, _pairs_schema, _read,
|
||
_relation_conflict, _root, _supplement_schema, _text_sections, _umbrella_schema,
|
||
_aspect_marker, _title_variants, _corpus_files, _evidence_pack, _sink_json, source_folder,
|
||
)
|
||
from config import (QA_GATE_NOTE, QA_GATE_LLM,
|
||
|
||
BLOCKS_GRUPPIERUNG_AKTIV, EMBEDDING_AKTIV, EMBEDDING_BLOCK_CAP,
|
||
EMBEDDING_SIBLING_CAP, EMBEDDING_SIBLING_FLOOR, FRAGMENT_MIN_COS,
|
||
GROUP_MIN_COS_FLOOR, GROUP_RECONCILE_FLOOR,
|
||
)
|
||
from fsutil import atomic_write_json, atomic_write_text
|
||
from jsonio import read_json_file as _json_file
|
||
from pipeline import (CANCELLED, FAILED, OK, GenContext, _extra, _log, _prompt,
|
||
_runde_schema, _timeout, _yesno_schema, run_single_slot)
|
||
from textkit import _norm_title, _parse_selection, _title, clean_title
|
||
|
||
log = logging.getLogger("creator.board_inventory")
|
||
|
||
BOARD = "inventory"
|
||
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 = clean_title(_title(record))
|
||
norm = _norm_title(title)
|
||
if not norm or norm in seen:
|
||
continue
|
||
seen.add(norm) # one reader = one vote per concept
|
||
parts = [t.strip() for t in record.split(" — ")]
|
||
desc = clean_title(parts[1]) if len(parts) >= 2 else ""
|
||
src = source or (parts[2] if len(parts) >= 3 else "")
|
||
async with _ingest_lock:
|
||
if await db.kanban_add_title(flow.topic, BOARD, norm, title, desc, src, reader):
|
||
n += 1
|
||
if n:
|
||
flow.wake.set()
|
||
return n
|
||
|
||
|
||
async def _research_once(ctx: GenContext, flow: Flow, q: dict, folder, instructions: str,
|
||
tag: str, *, section: str = "", fokus: str = "", source_file: str = ""):
|
||
"""ONE agent searches; its titles stream into the ingest queue LIVE. Two sources feed
|
||
the ingest: the JSON event stream (on_line → text buffer) AND the file the agent writes —
|
||
whichever the agent uses, cards stream in immediately (not only after it finishes)."""
|
||
work_dir = flow.work_dir
|
||
caps = "files" if folder else "full"
|
||
p = work_dir / f"research-{tag}.md"
|
||
stop = asyncio.Event()
|
||
buf: list[str] = [] # assistant text streamed live from the JSON events
|
||
|
||
def _on_line(raw: str): # sync, called per stdout line by the agent runner
|
||
if (t := _extract_text(raw)):
|
||
buf.append(t)
|
||
|
||
async def _drain() -> bool: # ingest from BOTH event buffer and file (idempotent)
|
||
text = "".join(buf)
|
||
if (ft := _file_payload(p)):
|
||
text += "\n" + ft
|
||
return bool(text) and await _ingest_titles(flow, text, tag, source_file)
|
||
|
||
async def _tail(): # live-ingest loop while the agent runs
|
||
while not stop.is_set():
|
||
try:
|
||
await asyncio.wait_for(stop.wait(), timeout=_POLL_RESEARCH)
|
||
except asyncio.TimeoutError:
|
||
pass
|
||
await _drain()
|
||
|
||
if not _file_payload(p): # resume: a valid reader file is re-ingested without an agent
|
||
p.unlink(missing_ok=True)
|
||
tail = asyncio.create_task(_tail())
|
||
try:
|
||
await run_single_slot(
|
||
ctx, f"Research {tag}", key=f"blocks-{ctx.topic}-research-{tag}",
|
||
prompt=_build_research_prompt(ctx.topic, p, instructions, q["type"], folder,
|
||
fokus=fokus, section=section),
|
||
role="quick", capabilities=caps,
|
||
payload=(lambda result, p=p: _file_payload(p)),
|
||
timeout=RESEARCH_RUNTIME, on_line=_on_line,
|
||
)
|
||
finally:
|
||
stop.set()
|
||
await tail
|
||
await _drain() # final catch-up (also the whole resume path)
|
||
_log(ctx.topic, f"Research {tag}: Titel → ingest")
|
||
|
||
|
||
def _build_producers(ctx: GenContext, flow: Flow, q: dict, folder, instructions: str) -> list:
|
||
"""Producer coroutines per source mode. thema = N web agents; uni/projekt = 2 readers per
|
||
text section; link = 2 readers per fixed page batch (coverage ticked per batch)."""
|
||
topic = ctx.topic
|
||
|
||
if not folder: # thema: free web research
|
||
return [_research_once(ctx, flow, q, folder, instructions, str(i))
|
||
for i in range(1, RESEARCH_THEMA_AGENTS + 1)]
|
||
|
||
if q["type"] in ("uni", "projekt"):
|
||
sections: list[tuple[str, str]] = []
|
||
for fn in sorted(set(_crawl_index(folder).values()) or
|
||
[f.name for f in sorted(folder.glob("**/*.txt"))]):
|
||
for sec in _text_sections(_read(folder / fn)):
|
||
sections.append((fn, sec))
|
||
prods = []
|
||
for ei, (fn, sec) in enumerate(sections, 1):
|
||
block = (f"ARBEITE AUSSCHLIESSLICH MIT DIESEM TEXTABSCHNITT. Lies ihn VOLLSTÄNDIG, "
|
||
f"überspringe nichts. Suche NICHT im Web — nur diese Section zählt."
|
||
f"\n\n-----\n{sec}\n-----")
|
||
prods += [_research_once(ctx, flow, q, folder, instructions, f"a{ei}-{r}",
|
||
section=block, source_file=fn)
|
||
for r in range(1, RESEARCH_READERS + 1)]
|
||
return prods
|
||
|
||
# link/crawl: fixed page batches, RESEARCH_READERS readers each
|
||
async def _batch(bi: int, batch: list[str]):
|
||
liste = "\n".join(f"- {p}" for p in batch)
|
||
fokus = ("WICHTIG — feste Assignment: Bearbeite AUSSCHLIESSLICH diese Dateien und lies JEDE "
|
||
f"vollständig. Ignoriere alle anderen Dateien im Ordner:\n{liste}")
|
||
try:
|
||
await asyncio.gather(*[
|
||
_research_once(ctx, flow, q, folder, instructions, f"b{bi}-{r}", fokus=fokus)
|
||
for r in range(1, RESEARCH_READERS + 1)], return_exceptions=True)
|
||
finally:
|
||
await db.mark_sources_read_done(topic, batch) # tick coverage even without hits
|
||
|
||
return [_batch(bi, b) for bi, b in enumerate(
|
||
_chunk_nums(sorted(flow.state["pages"]),
|
||
max(1, math.ceil(len(flow.state["pages"]) / RESEARCH_BATCH))), 1)]
|
||
|
||
|
||
# ── Column processors ──────────────────────────────────────────────────────────────
|
||
async def _proc_ingest(flow: Flow, cards):
|
||
"""Visibility column — exact dedup already folded at DB-add. Just advance."""
|
||
await db.kanban_advance_many(flow.topic, BOARD, [(c["card_id"], "cluster") for c in cards])
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_cluster(ctx: GenContext, flow: Flow, cards):
|
||
"""SERIAL online clustering: nearest neighbour over ALL known titles (including the
|
||
batch mates registered a moment ago — fixes main's same-batch blindness). Join the
|
||
neighbour's cluster (cos ≥ floor, cap 25) or open a new one; the touched cluster is
|
||
reworked from pair_check."""
|
||
topic = flow.topic
|
||
emb = await _emb_ok(flow)
|
||
membership = await db.kanban_membership(topic)
|
||
sizes: dict[str, int] = {}
|
||
for g in membership.values():
|
||
sizes[g] = sizes.get(g, 0) + 1
|
||
nn = flow.state.setdefault("nn", {"norms": [], "texts": []})
|
||
moves = []
|
||
for c in cards:
|
||
nm, p = c["card_id"], c["payload"]
|
||
text = _t_text(p)
|
||
target = None
|
||
if emb and nn["norms"]:
|
||
mat = await _vec_rows(flow, nn["texts"] + [text])
|
||
if mat is not None:
|
||
import numpy as np
|
||
cos = mat[:-1] @ mat[-1]
|
||
best = int(np.argmax(cos))
|
||
if float(cos[best]) >= DEDUP_PAIR_FLOOR:
|
||
cand = membership.get(nn["norms"][best])
|
||
if cand and sizes.get(cand, 0) < EMBEDDING_BLOCK_CAP:
|
||
target = cand
|
||
cid = target or f"cl-{uuid.uuid4().hex[:12]}"
|
||
membership[nm] = cid
|
||
sizes[cid] = sizes.get(cid, 0) + 1
|
||
await db.kanban_set_member(topic, nm, cid)
|
||
# gaining a member reworks the cluster from pair_check (live re-clustering)
|
||
await db.kanban_upsert_card(topic, BOARD, cid, "cluster", "pair_check",
|
||
{"title": p["title"], "description": p.get("description", "")})
|
||
nn["norms"].append(nm)
|
||
nn["texts"].append(text)
|
||
moves.append((nm, "clustered"))
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _member_rows(topic: str, cid: str) -> list[dict]:
|
||
rows = []
|
||
for nm in await db.kanban_members_of(topic, cid):
|
||
tc = await db.kanban_get_card(topic, BOARD, nm)
|
||
if tc:
|
||
rows.append({"norm": nm, **tc["payload"]})
|
||
return rows
|
||
|
||
|
||
async def _proc_pair_check(ctx: GenContext, flow: Flow, cards):
|
||
results = await asyncio.gather(*[_pair_one(ctx, flow, c) for c in cards], return_exceptions=True)
|
||
errs = [r for r in results if isinstance(r, Exception)]
|
||
if errs:
|
||
raise errs[0] # unadvanced cards go to backoff (engine)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _pair_one(ctx: GenContext, flow: Flow, c):
|
||
"""Pairwise entity resolution inside one cluster: candidate pairs = mean(title, title+desc)
|
||
cosine ≥ floor ∪ canonical-key blocking; a judge confirms each pair; auto-merges (title ≥0.95,
|
||
exact key); relation guard blocks different reductions; complete-link cliques merge — the
|
||
rest splits into fresh clusters. All resulting clusters → consensus_gate."""
|
||
topic = flow.topic
|
||
cid = c["card_id"]
|
||
rows = await _member_rows(topic, cid)
|
||
if len(rows) <= 1:
|
||
await db.kanban_advance(topic, BOARD, cid, "consensus_gate")
|
||
return
|
||
n = len(rows)
|
||
texts_full = [_t_text(r) for r in rows]
|
||
texts_title = [r["title"] for r in rows]
|
||
vf = await _vec_rows(flow, texts_full)
|
||
vt = await _vec_rows(flow, texts_title)
|
||
sims_title = vt @ vt.T if vt is not None else None
|
||
pairs: set[tuple[int, int]] = set()
|
||
if vf is not None and vt is not None:
|
||
sims = (vf @ vf.T + sims_title) / 2
|
||
pairs |= {(i, j) for i in range(n) for j in range(i + 1, n)
|
||
if float(sims[i][j]) >= DEDUP_PAIR_FLOOR}
|
||
else: # no model → judge all pairs (clusters are small)
|
||
pairs |= {(i, j) for i in range(n) for j in range(i + 1, n)}
|
||
key_groups: dict[str, list[int]] = {}
|
||
for i, r in enumerate(rows):
|
||
if (k := _canonical_key(r["title"])):
|
||
key_groups.setdefault(k, []).append(i)
|
||
for grp in key_groups.values():
|
||
for x in range(len(grp)):
|
||
for y in range(x + 1, len(grp)):
|
||
pairs.add((grp[x], grp[y]))
|
||
ordered = sorted(pairs)
|
||
edges: list[tuple[int, int]] = []
|
||
chunks = [ordered[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(ordered), DEDUP_PAIRS_CHUNK)]
|
||
h = _h(*[r["norm"] for r in rows])
|
||
|
||
async def _judge_pairs(ci, chunk): # chunks are independent — one parallel wave
|
||
path = flow.work_dir / f"pair-{cid}-{h}-c{ci}.json"
|
||
if _pairs_schema(_json_file(path)) is not None:
|
||
return # resume
|
||
lines = "\n\n".join(
|
||
f"{j + 1}.\nA: {_t_text(rows[a])}\nB: {_t_text(rows[b])}"
|
||
for j, (a, b) in enumerate(chunk))
|
||
status, _v = await run_single_slot(
|
||
ctx, f"Paar-Check {cid}", key=f"blocks-{topic}-pair-{cid}-{h}-c{ci}",
|
||
prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _pairs_schema(_json_file(p)),
|
||
timeout=_timeout("selection_mapping", len(chunk)))
|
||
if status == FAILED:
|
||
raise RuntimeError(f"Paar-Filter {cid} chunk {ci} ohne Ergebnis")
|
||
|
||
results = await asyncio.gather(*[_judge_pairs(ci, c) for ci, c in enumerate(chunks)],
|
||
return_exceptions=True)
|
||
errs = [r for r in results if isinstance(r, Exception)]
|
||
if errs:
|
||
raise errs[0]
|
||
if ctx.is_cancelled():
|
||
return
|
||
for ci, chunk in enumerate(chunks):
|
||
verdict = _pairs_schema(_json_file(flow.work_dir / f"pair-{cid}-{h}-c{ci}.json")) or {}
|
||
for j, (a, b) in enumerate(chunk):
|
||
if verdict.get(j + 1) and not _relation_conflict(rows[a]["title"], rows[b]["title"]):
|
||
edges.append((a, b))
|
||
if sims_title is not None: # auto recall net: near-identical titles merge without the judge
|
||
for a, b in ordered:
|
||
if float(sims_title[a][b]) >= DEDUP_TITLE_AUTO and not _relation_conflict(
|
||
rows[a]["title"], rows[b]["title"]):
|
||
edges.append((a, b))
|
||
for grp in key_groups.values(): # exact-canonical-key auto-merge (ER ~100% precision)
|
||
for x in range(len(grp)):
|
||
for y in range(x + 1, len(grp)):
|
||
a, b = grp[x], grp[y]
|
||
if not _relation_conflict(rows[a]["title"], rows[b]["title"]):
|
||
edges.append((a, b))
|
||
groups = _cliques(n, edges)
|
||
used = set().union(*[set(g) for g in groups]) if groups else set()
|
||
groups += [[i] for i in range(n) if i not in used]
|
||
groups.sort(key=len, reverse=True)
|
||
for gi, g in enumerate(groups):
|
||
gid = cid if gi == 0 else f"cl-{uuid.uuid4().hex[:12]}"
|
||
await db.kanban_set_members(topic, gid, [rows[i]["norm"] for i in g])
|
||
rep = rows[max(g, key=lambda k: len(rows[k].get("description") or ""))]
|
||
await db.kanban_upsert_card(topic, BOARD, gid, "cluster", "consensus_gate",
|
||
{"title": rep["title"], "description": rep.get("description", "")})
|
||
|
||
|
||
def _rep(rows: list[dict]) -> dict:
|
||
"""Cluster representative via the survivorship cascade (main concept, not an aspect)."""
|
||
cands = [{"title": r["title"], "description": r.get("description") or "",
|
||
"reader": r.get("readers") or []} for r in rows]
|
||
return _canonical(cands, list(range(len(cands))), set())
|
||
|
||
|
||
_ANKER_STOP = {"der", "die", "das", "und", "oder", "für", "mit", "von", "des", "den", "dem",
|
||
"ein", "eine", "einer", "the", "and", "for", "problem", "probleme",
|
||
"algorithmus", "algorithmen", "definition", "satz", "lemma", "beispiel",
|
||
"methode", "verfahren"}
|
||
|
||
|
||
def _korpus_tokens(folder) -> set[str]:
|
||
"""All corpus word tokens (≥3 chars, casefolded) — anchor base for the reader-title gate."""
|
||
toks: set[str] = set()
|
||
for f in _corpus_files(folder, None):
|
||
try:
|
||
toks |= set(re.findall(r"\w{3,}", f.read_text(encoding="utf-8").casefold()))
|
||
except OSError:
|
||
continue
|
||
return toks
|
||
|
||
|
||
def _hat_anker(title: str, ctoks: set[str]) -> bool:
|
||
"""≥1 distinctive title token appears in the corpus — digit-suffix tolerant:
|
||
'∆TSP1' → 'tsp1' → 'tsp' (the corpus tokenizes '∆TSP' to 'tsp')."""
|
||
for t in re.findall(r"\w{3,}", title.casefold()):
|
||
if t in _ANKER_STOP:
|
||
continue
|
||
forms = {t}
|
||
a = unicodedata.normalize("NFKD", t).encode("ascii", "ignore").decode()
|
||
if len(a) >= 3:
|
||
forms.add(a) # Symbol-Präfixe (δtsp1 → tsp1) — der Korpus-Tokenizer kennt kein ∆
|
||
forms |= {f2 for f in list(forms) if len(f2 := f.rstrip("0123456789")) >= 3}
|
||
if any(ct == f or ct.startswith(f) for f in forms for ct in ctoks):
|
||
return True
|
||
return False
|
||
|
||
|
||
async def _anker_beleg(ctx: GenContext, flow: Flow, kandidaten: list[tuple[str, dict]]) -> set[str]:
|
||
"""Evidence judge for quorum titles WITHOUT any corpus anchor — two readers naming the
|
||
same famous canon independently beat the quorum although the material never mentions it
|
||
(measured: 'Königsberger Brückenproblem', 0 corpus hits). FAIL-OPEN: the titles carry
|
||
2-reader backing, only an explicit 'nein' rejects. → card_ids to reject."""
|
||
topic = flow.topic
|
||
folder = source_folder(topic)
|
||
lines = []
|
||
for k, (cid, p) in enumerate(kandidaten, 1):
|
||
ev = _evidence_pack(folder, None, [p.get("title", ""), p.get("description") or ""], budget=4000)
|
||
lines.append(f"{k}. {p.get('title', '')} — {p.get('description') or ''}\nAUSZÜGE:\n"
|
||
f"{ev or '(keine passenden Auszüge im Material gefunden)'}")
|
||
h = _h(*[cid for cid, _ in kandidaten])
|
||
path = flow.work_dir / f"anker-beleg-{h}.json"
|
||
ids = set(range(1, len(kandidaten) + 1))
|
||
verdict = _yesno_schema(_json_file(path), ids)
|
||
if verdict is None:
|
||
status, verdict = await run_single_slot(
|
||
ctx, "Anker-Beleg", key=f"blocks-{topic}-anker-beleg-{h}",
|
||
prompt=_prompt("Blocks-Supplement-Beleg", topic=topic, proposals="\n\n".join(lines),
|
||
extra=_extra(flow.state.get("instructions", ""))),
|
||
role="judge", capabilities="none",
|
||
payload=lambda result, p2=path, i=ids: _sink_json(result, p2, lambda d: _yesno_schema(d, i)),
|
||
timeout=_timeout("selection_mapping", len(kandidaten)))
|
||
if status != OK or not isinstance(verdict, dict):
|
||
verdict = {} # fail-open
|
||
return {cid for k, (cid, _p) in enumerate(kandidaten, 1) if str(verdict.get(k, "ja")) == "nein"}
|
||
|
||
|
||
async def _proc_consensus_gate(ctx: GenContext, flow: Flow, cards):
|
||
"""Code gate: reader union ≥2 (or supplement) passes; single finds → clarify.
|
||
Reference-titled consensus clusters also go to clarify (majority quorum + rename).
|
||
uni/projekt: quorum titles without ANY corpus anchor face the evidence judge —
|
||
reader co-hallucination of famous canon beats the quorum otherwise."""
|
||
topic = flow.topic
|
||
moves = []
|
||
anker_kandidaten: list[tuple[str, dict]] = []
|
||
folder = source_folder(topic)
|
||
ctoks = None
|
||
if folder is not None:
|
||
ctoks = flow.state.get("korpus_tokens")
|
||
if ctoks is None:
|
||
ctoks = flow.state["korpus_tokens"] = await asyncio.to_thread(_korpus_tokens, folder)
|
||
for c in cards:
|
||
cid = c["card_id"]
|
||
rows = await _member_rows(topic, cid)
|
||
if not rows:
|
||
moves.append((cid, "done_cluster"))
|
||
continue
|
||
readers = set().union(*[set(r.get("readers") or []) for r in rows])
|
||
supplement = any(r.get("supplement") for r in rows)
|
||
rep = _rep(rows)
|
||
p = c["payload"]
|
||
p.update(title=rep["title"], description=rep["description"], readers=sorted(readers),
|
||
supplement=supplement, n_size=len(readers)) # LPT: big evidence first
|
||
if supplement or len(readers) >= 2:
|
||
if _is_reference(rep["title"]) and not supplement:
|
||
p["quorum"] = "majority" # consensus reference title: rename/exam, not the hard bar
|
||
moves.append((cid, "clarify"))
|
||
elif ctoks and not supplement and not _hat_anker(rep["title"], ctoks):
|
||
anker_kandidaten.append((cid, p))
|
||
else:
|
||
moves.append((cid, "naming"))
|
||
else:
|
||
moves.append((cid, "clarify"))
|
||
await db.kanban_set_payload(topic, BOARD, cid, p)
|
||
if anker_kandidaten:
|
||
weg = await _anker_beleg(ctx, flow, anker_kandidaten)
|
||
for cid, p in anker_kandidaten:
|
||
if cid in weg:
|
||
p["reason"] = "kein-beleg"
|
||
await db.kanban_set_payload(topic, BOARD, cid, p)
|
||
moves.append((cid, "rejected"))
|
||
else:
|
||
moves.append((cid, "naming"))
|
||
if weg:
|
||
_log(topic, f"Anker-Beleg: {len(weg)} Quorum-Titel ohne Materialbeleg verworfen")
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_clarify(ctx: GenContext, flow: Flow, cards):
|
||
"""Single-reader finds: deterministic pre-reject, then a 3-judge panel (Blocks-Klaerung).
|
||
Quorum: UNANIMITY for single-reader clusters (origin-split C1), majority for reference-titled
|
||
consensus clusters. Kept reference titles get the panel's rename."""
|
||
topic = flow.topic
|
||
moves: list[tuple[str, str]] = []
|
||
pending = []
|
||
for c in cards:
|
||
t = c["payload"].get("title", "")
|
||
d = c["payload"].get("description", "")
|
||
if (_is_artifact(t) or _FILTER_NOTATION.search(t)) and not _is_named_statement(t, d):
|
||
c["payload"]["reason"] = "pre-reject"
|
||
await db.kanban_set_payload(topic, BOARD, c["card_id"], c["payload"])
|
||
moves.append((c["card_id"], "rejected"))
|
||
else:
|
||
pending.append(c)
|
||
if pending:
|
||
h = _h(*[c["card_id"] for c in pending])
|
||
rest = "\n".join(f"- {_t_text(c['payload'])}" for c in pending)
|
||
judges = [(j, flow.work_dir / f"clarify-{h}-j{j}.json")
|
||
for j in range(1, CONSOLIDATION_PANEL + 1)]
|
||
|
||
async def _judge(j, path):
|
||
if _runde_schema(_json_file(path)) is not None:
|
||
return # resume: keep a valid judge file
|
||
await run_single_slot(
|
||
ctx, f"Klärung j{j}", key=f"blocks-{topic}-clarify-{h}-j{j}",
|
||
prompt=_prompt("Blocks-Klaerung", topic=topic, rest=rest,
|
||
final="\n- Entscheide JEDEN Eintrag. `rest` MUSS leer sein.",
|
||
out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _runde_schema(_json_file(p)),
|
||
timeout=_timeout("selection_mapping", len(pending)))
|
||
|
||
await asyncio.gather(*[_judge(j, p) for j, p in judges], return_exceptions=True)
|
||
if ctx.is_cancelled():
|
||
return
|
||
outs, raws = [], []
|
||
for _, path in judges:
|
||
raw = _json_file(path)
|
||
r = _runde_schema(raw)
|
||
if r is not None:
|
||
outs.append(r)
|
||
raws.append(raw)
|
||
if not outs:
|
||
raise RuntimeError("Klärung: kein Judge lieferte ein Ergebnis")
|
||
votes: dict[str, int] = {}
|
||
for accepted, _ in outs:
|
||
for nt in {_norm_title(_title(t)) for t in accepted}:
|
||
votes[nt] = votes.get(nt, 0) + 1
|
||
renames: dict[str, dict[str, int]] = {}
|
||
for raw in raws:
|
||
if isinstance(raw, dict) and isinstance(raw.get("rename"), dict):
|
||
for old, new in raw["rename"].items():
|
||
new = str(new).strip()
|
||
if new:
|
||
renames.setdefault(_norm_title(str(old)), {}).setdefault(new, 0)
|
||
renames[_norm_title(str(old))][new] += 1
|
||
for c in pending:
|
||
p = c["payload"]
|
||
norm = _norm_title(p.get("title", ""))
|
||
v = votes.get(norm, 0)
|
||
majority = p.get("quorum") == "majority"
|
||
accept = (v * 2 >= len(outs)) if majority else (v >= len(outs))
|
||
if not accept:
|
||
p.update(reason="failed-quorum", votes=v, judges=len(outs))
|
||
await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
|
||
moves.append((c["card_id"], "rejected"))
|
||
continue
|
||
if _is_reference(p.get("title", "")) and (sug := renames.get(norm)):
|
||
best = max(sug, key=lambda k: (sug[k], len(k)))
|
||
if not _is_reference(best):
|
||
p.update(title=best, renamed=True)
|
||
await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
|
||
moves.append((c["card_id"], "naming"))
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _choose_title(ctx: GenContext, flow: Flow, cid: str, rows: list[dict],
|
||
template: str, current: 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, "n_size": len(readers),
|
||
"supplement": bool(p.get("supplement")), "cluster": cid,
|
||
})
|
||
await db.kanban_advance(topic, BOARD, cid, "done_cluster")
|
||
|
||
|
||
# ── Block barriers ─────────────────────────────────────────────────────────────────
|
||
async def _context_blocks(topic: str, exclude: set[str]) -> list[dict]:
|
||
"""Blocks already past the filter (grouping/gap_check/done_block) — context for the
|
||
relational judges of a LATER pass (supplement feedback), never demotable themselves."""
|
||
out = []
|
||
for r in await db.kanban_cards(topic, board=BOARD, kind="block"):
|
||
if r["card_id"] in exclude or r["stage"] not in ("grouping", "gap_check", "done", "done_block"):
|
||
continue
|
||
p = r["payload"]
|
||
out.append({"card_id": r["card_id"], "payload": p,
|
||
"title": p.get("title", ""), "description": p.get("description") or "",
|
||
"title_norm": _norm_title(p.get("title", ""))})
|
||
return out
|
||
|
||
|
||
async def _proc_fragment_filter(ctx: GenContext, flow: Flow, cards):
|
||
"""BARRIER/drain — the global pass over ALL block cards in the stage:
|
||
(a) deterministic re-merge across cluster borders (canonical key + title-cos ≥0.95,
|
||
catches cap splits), complete-link + relation guard, survivorship keeps the champion;
|
||
(b) the degrade pass from blocks.py: full list per judge, hard-drop double gate,
|
||
containment demote, parentless-noise drop, recheck panel, statement-gate rescue,
|
||
transitive root resolution. Fragments/drops → rejected (journal), survivors → grouping."""
|
||
topic = flow.topic
|
||
work_dir = flow.work_dir
|
||
rows = [{"card_id": c["card_id"], "payload": c["payload"],
|
||
"title": c["payload"].get("title", ""),
|
||
"description": c["payload"].get("description") or "",
|
||
"title_norm": _norm_title(c["payload"].get("title", ""))} for c in cards]
|
||
moves: list[tuple[str, str]] = []
|
||
|
||
# (a) global re-merge
|
||
n = len(rows)
|
||
if n >= 2:
|
||
edges = []
|
||
vt = await _vec_rows(flow, [r["title"] for r in rows])
|
||
keys: dict[str, list[int]] = {}
|
||
for i, r in enumerate(rows):
|
||
if (k := _canonical_key(r["title"])):
|
||
keys.setdefault(k, []).append(i)
|
||
for grp in keys.values():
|
||
for x in range(len(grp)):
|
||
for y in range(x + 1, len(grp)):
|
||
a, b = grp[x], grp[y]
|
||
if not _relation_conflict(rows[a]["title"], rows[b]["title"]):
|
||
edges.append((a, b))
|
||
if vt is not None:
|
||
sims_t = vt @ vt.T
|
||
for i in range(n):
|
||
for j in range(i + 1, n):
|
||
if float(sims_t[i][j]) >= DEDUP_TITLE_AUTO and not _relation_conflict(
|
||
rows[i]["title"], rows[j]["title"]):
|
||
edges.append((i, j))
|
||
merged = 0
|
||
for g in _cliques(n, edges):
|
||
rep = max(g, key=lambda k: (-_aspect_marker(rows[k]["title"]),
|
||
len(rows[k]["description"]), len(rows[k]["title"]), -k))
|
||
rp = rows[rep]["payload"]
|
||
for k in g:
|
||
if k == rep:
|
||
continue
|
||
lp = rows[k]["payload"]
|
||
rp["readers"] = sorted(set(rp.get("readers") or []) | set(lp.get("readers") or []))
|
||
rp["sources"] = sorted(set(rp.get("sources") or []) | set(lp.get("sources") or []))
|
||
lp.update(reason="merged", merged_into=rows[rep]["title"])
|
||
await db.kanban_set_payload(topic, BOARD, rows[k]["card_id"], lp)
|
||
moves.append((rows[k]["card_id"], "grouped"))
|
||
merged += 1
|
||
await db.kanban_set_payload(topic, BOARD, rows[rep]["card_id"], rp)
|
||
if merged:
|
||
dropped = {cid for cid, _ in moves}
|
||
rows = [r for r in rows if r["card_id"] not in dropped]
|
||
_log(topic, f"Fragment-Filter: {merged} Cap-Split-Dublette(n) re-merged")
|
||
|
||
# (b) degrade pass — full list = demotable stage rows + already-confirmed context
|
||
context = await _context_blocks(topic, exclude={r["card_id"] for r in rows})
|
||
allrows = rows + context
|
||
n_dem, n_all = len(rows), len(allrows)
|
||
if n_dem:
|
||
def _fline(i):
|
||
r = allrows[i - 1]
|
||
mark = "⚠ " if i <= n_dem and _filter_suspect(r) else ""
|
||
tail = "" if i <= n_dem else " (bereits bestätigt)"
|
||
d = r.get("description")
|
||
return (f"{i}. {mark}{r['title']} — {d}{tail}" if d else f"{i}. {mark}{r['title']}{tail}")
|
||
|
||
full_list = "\n".join(_fline(i) for i in range(1, n_all + 1))
|
||
h = _h(*[r["card_id"] for r in rows])
|
||
chunks = [list(range(i, min(i + FILTER_CHUNK, n_dem + 1))) for i in range(1, n_dem + 1, FILTER_CHUNK)]
|
||
|
||
# ONE wave: all chunk judges in parallel (verdicts are independent; the voting/
|
||
# containment evaluation below reads the files strictly afterwards).
|
||
async def _judge_chunk(ci, numbers):
|
||
path = work_dir / f"filter-{h}-c{ci}.json"
|
||
if _filter_schema(_json_file(path)) is not None:
|
||
return # resume: keep a valid judge file
|
||
status, _v = await run_single_slot(
|
||
ctx, f"Fragment-Filter {ci}", key=f"blocks-{topic}-filter-{h}-c{ci}",
|
||
prompt=_prompt("Blocks-Filter", topic=topic, list=full_list,
|
||
from_n=numbers[0], to_n=numbers[-1], out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _filter_schema(_json_file(p)),
|
||
timeout=_timeout("selection_mapping", len(numbers)))
|
||
if status == FAILED:
|
||
raise RuntimeError(f"Fragment-Filter chunk {ci} ohne Ergebnis")
|
||
|
||
results = await asyncio.gather(*[_judge_chunk(ci, nm) for ci, nm in enumerate(chunks)],
|
||
return_exceptions=True)
|
||
errs = [r for r in results if isinstance(r, Exception)]
|
||
if errs:
|
||
raise errs[0]
|
||
if ctx.is_cancelled():
|
||
return
|
||
proposals: dict[int, int] = {} # judge demotes are PROPOSALS — panel/containment confirm
|
||
drops: set[int] = set()
|
||
for ci, numbers in enumerate(chunks):
|
||
raw = _json_file(work_dir / f"filter-{h}-c{ci}.json")
|
||
verdict = _filter_schema(raw) or {}
|
||
nset = set(numbers)
|
||
for nr, parent in verdict.items():
|
||
if 1 <= parent <= n_all and nr in nset:
|
||
proposals[nr] = parent
|
||
for x in (raw.get("drop", []) if isinstance(raw, dict) else []):
|
||
try:
|
||
dnr = int(x)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if dnr in nset:
|
||
drops.add(dnr)
|
||
# hard-drop double gate
|
||
honored = {nr for nr in drops if _is_artifact(allrows[nr - 1]["title"])}
|
||
for nr in honored:
|
||
proposals.pop(nr, None)
|
||
# containment demote (deterministic — auto-confirms proposals whose parent name is
|
||
# literally contained in the title, and keeps the legacy ⚠-only pass) + parentless noise
|
||
norms = [(i, allrows[i - 1]["title_norm"]) for i in range(1, n_all + 1)]
|
||
fragments: dict[int, int] = {}
|
||
contained: set[int] = set()
|
||
for i in range(1, n_dem + 1):
|
||
if i in honored or (i not in proposals and not _filter_suspect(allrows[i - 1])):
|
||
continue
|
||
parent = _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i])
|
||
if parent is not None and parent != i and parent not in honored:
|
||
fragments[i] = parent
|
||
contained.add(i)
|
||
proposals.pop(i, None)
|
||
for i in range(1, n_dem + 1):
|
||
if i in fragments or i in honored or i in proposals or not _is_parentless_noise(allrows[i - 1]["title"]):
|
||
continue
|
||
if _containment_parent(allrows[i - 1]["title_norm"], [(nr, t) for nr, t in norms if nr != i]) is None:
|
||
honored.add(i)
|
||
# recheck panel = second opinion: unconfirmed judge proposals (WITHOUT the suggested
|
||
# parent — no anchoring) plus still-⚠ survivors. Majority ≥2 demotes/drops; a proposal
|
||
# the panel does not confirm survives. The panel is load-bearing now: fewer than 2
|
||
# valid judge files per chunk is an error (backoff), not a silent keep.
|
||
survivors = [i for i in range(1, n_dem + 1)
|
||
if i not in fragments and i not in honored
|
||
and (i in proposals or _filter_suspect(allrows[i - 1]))]
|
||
overruled: list[int] = []
|
||
if survivors:
|
||
ph = _h(",".join(map(str, survivors))) # panel-input hash: stale pre-change files never match
|
||
rchunks = [survivors[k:k + FILTER_CHUNK] for k in range(0, len(survivors), FILTER_CHUNK)]
|
||
|
||
async def _recheck_judge(ci, nums, j):
|
||
path = work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json"
|
||
if _filter_schema(_json_file(path)) is not None:
|
||
return # resume
|
||
await run_single_slot(
|
||
ctx, f"Filter-Recheck {ci}/{j}", key=f"blocks-{topic}-filter-recheck-{h}-{ph}-c{ci}-j{j}",
|
||
prompt=_prompt("Blocks-Filter-Recheck", topic=topic,
|
||
survivors="\n".join(_fline(i) for i in nums),
|
||
list=full_list, out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _filter_schema(_json_file(p)),
|
||
timeout=_timeout("selection_mapping", len(nums)))
|
||
|
||
await asyncio.gather(*[
|
||
_recheck_judge(ci, nums, j)
|
||
for ci, nums in enumerate(rchunks) for j in range(1, FILTER_RECHECK_PANEL + 1)],
|
||
return_exceptions=True)
|
||
if ctx.is_cancelled():
|
||
return
|
||
for ci, nums in enumerate(rchunks):
|
||
dem: dict[int, list[int]] = {}
|
||
drp: dict[int, int] = {}
|
||
nset = set(nums)
|
||
valid = 0
|
||
for j in range(1, FILTER_RECHECK_PANEL + 1):
|
||
raw = _json_file(work_dir / f"filter-recheck-{h}-{ph}-c{ci}-j{j}.json")
|
||
v = _filter_schema(raw)
|
||
if v is None:
|
||
continue
|
||
valid += 1
|
||
for nr, parent in v.items():
|
||
if nr in nset and 1 <= parent <= n_all and nr != parent:
|
||
dem.setdefault(nr, []).append(parent)
|
||
for x in (raw.get("drop", []) if isinstance(raw, dict) else []):
|
||
try:
|
||
dnr = int(x)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if dnr in nset:
|
||
drp[dnr] = drp.get(dnr, 0) + 1
|
||
if valid < 2:
|
||
raise RuntimeError(f"Filter-Recheck chunk {ci}: nur {valid} Judge(s) mit Ergebnis")
|
||
for nr in nums:
|
||
if nr in fragments or nr in honored:
|
||
continue
|
||
if drp.get(nr, 0) >= 2 and (_is_artifact(allrows[nr - 1]["title"])
|
||
or _is_parentless_noise(allrows[nr - 1]["title"])):
|
||
honored.add(nr)
|
||
elif len(dem.get(nr, [])) >= 2:
|
||
fragments[nr] = max(set(dem[nr]), key=dem[nr].count)
|
||
elif nr in proposals:
|
||
overruled.append(nr)
|
||
# embedding backstop: veto confirmed non-containment demotes whose direct title pair is
|
||
# literally structureless (see FRAGMENT_MIN_COS) — applied BEFORE _root resolution.
|
||
floor_veto: list[int] = []
|
||
if (cand := [nr for nr in fragments if nr not in contained]):
|
||
va = await _vec_rows(flow, [r["title"] for r in allrows])
|
||
if va is not None:
|
||
for nr in cand:
|
||
if float(va[nr - 1] @ va[fragments[nr] - 1]) < FRAGMENT_MIN_COS:
|
||
fragments.pop(nr)
|
||
floor_veto.append(nr)
|
||
# statement-gate rescue (final override)
|
||
def _protected(nr):
|
||
return _is_named_statement(allrows[nr - 1]["title"], allrows[nr - 1]["description"])
|
||
for nr in [nr for nr in list(fragments) if _protected(nr)]:
|
||
fragments.pop(nr, None)
|
||
honored -= {nr for nr in honored if _protected(nr)}
|
||
# transitive resolution + moves
|
||
journal = []
|
||
for nr in sorted(fragments):
|
||
root, cyclic = _root(nr, fragments)
|
||
if cyclic or not (1 <= root <= n_all) or nr > n_dem:
|
||
continue
|
||
r = allrows[nr - 1]
|
||
parent_norm = None if root in honored else allrows[root - 1]["title_norm"]
|
||
r["payload"].update(reason="fragment" if parent_norm else "drop-collateral",
|
||
parent_norm=parent_norm)
|
||
await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
|
||
moves.append((r["card_id"], "rejected"))
|
||
journal.append({"fragment": r["title"],
|
||
"eltern": allrows[root - 1]["title"] if parent_norm else None})
|
||
for nr in sorted(honored):
|
||
if nr > n_dem:
|
||
continue
|
||
r = allrows[nr - 1]
|
||
r["payload"].update(reason="drop", parent_norm=None)
|
||
await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
|
||
moves.append((r["card_id"], "rejected"))
|
||
journal.append({"fragment": r["title"], "eltern": None, "grund": "drop"})
|
||
# one journal file per pass (h) — the supplement feedback pass must not overwrite
|
||
# the main pass's journal (it is the evaluation instrument).
|
||
atomic_write_json(work_dir / f"inventar-filter-{h}.json",
|
||
{"vorher": n_dem, "degradiert": len(journal), "fragments": journal,
|
||
"ueberstimmt": [allrows[nr - 1]["title"] for nr in overruled],
|
||
"floor_veto": [allrows[nr - 1]["title"] for nr in floor_veto]}, indent=1)
|
||
_log(topic, f"Fragment-Filter: {n_dem} → {n_dem - len(journal)} (−{len(journal)})")
|
||
demoted = {cid for cid, _ in moves}
|
||
moves += [(r["card_id"], "dedup") for r in rows if r["card_id"] not in demoted]
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_dedup(ctx: GenContext, flow: Flow, cards):
|
||
"""BARRIER/drain — global pair dedup over the NAMED blocks: filter's re-merge only
|
||
catches key-exact/≥0.95 titles inside the stage; here embedding candidates (mean OR
|
||
title cosine ≥ DEDUP_GLOBAL_FLOOR) + canonical-key blocking go to a TWO-judge panel
|
||
(merge needs unanimity — single judges conflate variants with their base entity),
|
||
auto edges (title ≥0.95, exact key) with relation guard, complete-link cliques merge
|
||
into the champion (readers/sources union). The second wave (supplement) compares
|
||
against the already-confirmed context blocks. Journal carries every pair verdict."""
|
||
topic = flow.topic
|
||
work_dir = flow.work_dir
|
||
rows = [{"card_id": c["card_id"], "payload": c["payload"],
|
||
"title": c["payload"].get("title", ""),
|
||
"description": c["payload"].get("description") or ""} for c in cards]
|
||
context = await _context_blocks(topic, exclude={r["card_id"] for r in rows})
|
||
allrows = rows + context
|
||
n_dem, n_all = len(rows), len(allrows)
|
||
|
||
async def _pass_through():
|
||
await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "grouping") for r in rows])
|
||
flow.wake.set()
|
||
|
||
if n_all < 2 or not await _emb_ok(flow):
|
||
await _pass_through()
|
||
return
|
||
vf = await _vec_rows(flow, [_t_text(r) for r in allrows])
|
||
# titles casefolded: ALL-CAPS variants ("VERTEX COVER" vs "Vertex Cover (VC)") tank the
|
||
# raw title cosine below the candidate floor
|
||
vt = await _vec_rows(flow, [r["title"].casefold() for r in allrows])
|
||
if vf is None or vt is None:
|
||
await _pass_through()
|
||
return
|
||
sims_title = vt @ vt.T
|
||
sims = (vf @ vf.T + sims_title) / 2
|
||
|
||
def _demotable_pair(a: int, b: int) -> bool:
|
||
return a < n_dem or b < n_dem # confirmed context blocks never merge among themselves
|
||
|
||
pairs: set[tuple[int, int]] = set()
|
||
for i in range(n_all):
|
||
for j in range(i + 1, n_all):
|
||
# title-only cosine as second candidate source: descriptions of the same
|
||
# entity often stress different facets and dilute the mean below the floor
|
||
if _demotable_pair(i, j) and (float(sims[i][j]) >= DEDUP_GLOBAL_FLOOR
|
||
or float(sims_title[i][j]) >= DEDUP_GLOBAL_FLOOR):
|
||
pairs.add((i, j))
|
||
keys: dict[str, list[int]] = {}
|
||
for i, r in enumerate(allrows):
|
||
if (k := _canonical_key(r["title"])):
|
||
keys.setdefault(k, []).append(i)
|
||
for grp in keys.values():
|
||
for x in range(len(grp)):
|
||
for y in range(x + 1, len(grp)):
|
||
if _demotable_pair(grp[x], grp[y]):
|
||
pairs.add((grp[x], grp[y]))
|
||
# acronym↔expansion pairs ("SAT" vs "Satisfiability Problem (SAT)"): the title cosine
|
||
# of short vs long form sits far below the floor — variant match makes them candidates,
|
||
# the judge panel decides as usual (no auto-merge)
|
||
idents = [{_norm_title(r["title"]), _canonical_key(r["title"])} - {""} for r in allrows]
|
||
variants = [_title_variants(r["title"]) for r in allrows]
|
||
for i in range(n_all):
|
||
for j in range(i + 1, n_all):
|
||
if _demotable_pair(i, j) and (variants[i] & idents[j] or variants[j] & idents[i]):
|
||
pairs.add((i, j))
|
||
ordered = sorted(pairs)
|
||
h = _h(*[r["card_id"] for r in allrows])
|
||
if not ordered:
|
||
atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
|
||
{"vorher": n_dem, "paare": 0, "merged": []}, indent=1)
|
||
await _pass_through()
|
||
return
|
||
chunks = [ordered[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(ordered), DEDUP_PAIRS_CHUNK)]
|
||
|
||
async def _judge(ci, chunk, jj):
|
||
path = work_dir / f"dedup-{h}-c{ci}-j{jj}.json"
|
||
if _pairs_schema(_json_file(path)) is not None:
|
||
return # resume
|
||
lines = "\n\n".join(f"{j + 1}.\nA: {_t_text(allrows[a])}\nB: {_t_text(allrows[b])}"
|
||
for j, (a, b) in enumerate(chunk))
|
||
status, _v = await run_single_slot(
|
||
ctx, f"Dedup {ci} j{jj}", key=f"blocks-{topic}-dedup-{h}-c{ci}-j{jj}",
|
||
prompt=_prompt("Blocks-Dedup", topic=topic, pairs=lines, out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _pairs_schema(_json_file(p)),
|
||
timeout=_timeout("selection_mapping", len(chunk)))
|
||
if status == FAILED:
|
||
raise RuntimeError(f"Dedup chunk {ci} j{jj} ohne Ergebnis")
|
||
|
||
# two-judge panel per chunk (one wave): a merge needs UNANIMITY — the observed
|
||
# failure mode is a single judge conflating a variant with its base entity
|
||
results = await asyncio.gather(*[_judge(ci, c, jj) for ci, c in enumerate(chunks)
|
||
for jj in (1, 2)], return_exceptions=True)
|
||
errs = [r for r in results if isinstance(r, Exception)]
|
||
if errs:
|
||
raise errs[0]
|
||
if ctx.is_cancelled():
|
||
return
|
||
edges: list[tuple[int, int]] = []
|
||
detail: dict[tuple[int, int], str] = {}
|
||
|
||
def _edge(a, b, kanal):
|
||
if _relation_conflict(allrows[a]["title"], allrows[b]["title"]):
|
||
detail[(a, b)] = "guard_veto"
|
||
else:
|
||
edges.append((a, b))
|
||
detail[(a, b)] = kanal
|
||
|
||
for ci, chunk in enumerate(chunks):
|
||
v1 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j1.json")) or {}
|
||
v2 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j2.json")) or {}
|
||
for j, (a, b) in enumerate(chunk):
|
||
ja1, ja2 = bool(v1.get(j + 1)), bool(v2.get(j + 1))
|
||
if ja1 and ja2:
|
||
_edge(a, b, "ja")
|
||
else:
|
||
detail[(a, b)] = "nein" if not (ja1 or ja2) else "uneinig"
|
||
for a, b in ordered: # auto recall net: near-identical titles merge without the judges
|
||
if float(sims_title[a][b]) >= DEDUP_TITLE_AUTO:
|
||
_edge(a, b, "auto_titel")
|
||
for grp in keys.values(): # exact-canonical-key auto-merge
|
||
for x in range(len(grp)):
|
||
for y in range(x + 1, len(grp)):
|
||
if _demotable_pair(grp[x], grp[y]):
|
||
_edge(grp[x], grp[y], "auto_key")
|
||
moves: list[tuple[str, str]] = []
|
||
journal: list[dict] = []
|
||
gone: set[int] = set()
|
||
for g in _cliques(n_all, edges):
|
||
ctx_members = [k for k in g if k >= n_dem]
|
||
if ctx_members:
|
||
rep = ctx_members[0] # confirmed block always wins
|
||
else:
|
||
rep = max(g, key=lambda k: (-_aspect_marker(allrows[k]["title"]),
|
||
len(allrows[k]["description"]), len(allrows[k]["title"]), -k))
|
||
rp = allrows[rep]["payload"]
|
||
changed = False
|
||
for k in g:
|
||
if k == rep or k >= n_dem:
|
||
continue # context members stay untouched
|
||
lp = allrows[k]["payload"]
|
||
rp["readers"] = sorted(set(rp.get("readers") or []) | set(lp.get("readers") or []))
|
||
rp["sources"] = sorted(set(rp.get("sources") or []) | set(lp.get("sources") or []))
|
||
lp.update(reason="merged", merged_into=allrows[rep]["title"])
|
||
await db.kanban_set_payload(topic, BOARD, allrows[k]["card_id"], lp)
|
||
moves.append((allrows[k]["card_id"], "grouped"))
|
||
gone.add(k)
|
||
journal.append({"dublette": allrows[k]["title"], "in": allrows[rep]["title"]})
|
||
changed = True
|
||
if changed:
|
||
await db.kanban_set_payload(topic, BOARD, allrows[rep]["card_id"], rp)
|
||
if rep >= n_dem and rp.get("mirrored_norm"): # mirror refresh (sources only)
|
||
await db.upsert_block(topic, rp["mirrored_norm"], allrows[rep]["title"],
|
||
allrows[rep]["description"], rp.get("sources") or [])
|
||
moves += [(r["card_id"], "grouping") for i, r in enumerate(rows) if i not in gone]
|
||
atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
|
||
{"vorher": n_dem, "paare": len(ordered), "merged": journal,
|
||
"paare_detail": [{"a": allrows[a]["title"], "b": allrows[b]["title"],
|
||
"verdict": v} for (a, b), v in sorted(detail.items())]},
|
||
indent=1)
|
||
if journal:
|
||
_log(topic, f"Dedup: {len(journal)} Dublette(n) zusammengelegt")
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_grouping(ctx: GenContext, flow: Flow, cards):
|
||
"""BARRIER/drain — umbrella grouping over the filter survivors: embedding sibling clusters
|
||
(low floor, high recall) + top-down pass, one judge per cluster, type gate + min-cos backstop,
|
||
reconcile pass, completion judge. Members → grouped; the umbrella becomes a new block card."""
|
||
topic = flow.topic
|
||
work_dir = flow.work_dir
|
||
rows = [{"card_id": c["card_id"], "payload": c["payload"],
|
||
"title": c["payload"].get("title", ""),
|
||
"description": c["payload"].get("description") or "",
|
||
"title_norm": _norm_title(c["payload"].get("title", ""))} for c in cards]
|
||
n = len(rows)
|
||
if not (BLOCKS_GRUPPIERUNG_AKTIV and await _emb_ok(flow)) or n < 3:
|
||
await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "gap_check") for r in rows])
|
||
flow.wake.set()
|
||
return
|
||
texts = [_t_text(r) for r in rows]
|
||
vv = await _vec_rows(flow, texts)
|
||
if vv is None:
|
||
await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "gap_check") for r in rows])
|
||
flow.wake.set()
|
||
return
|
||
sims = vv @ vv.T
|
||
clusters = await asyncio.to_thread(embedding.capped_blocks, sims,
|
||
EMBEDDING_SIBLING_FLOOR, EMBEDDING_SIBLING_CAP)
|
||
multi = [c for c in clusters if len(c) > 1]
|
||
all_ids = set(range(1, n + 1))
|
||
full_list = "\n".join(f"{i}. {texts[i - 1]}" for i in range(1, n + 1))
|
||
h = _h(*[r["card_id"] for r in rows])
|
||
|
||
def _min_cos(idxs):
|
||
if len(idxs) < 2:
|
||
return 1.0
|
||
return round(min(float(sims[i][j]) for a, i in enumerate(idxs) for j in idxs[a + 1:]), 3)
|
||
|
||
async def _assess(tag, cand_text, count):
|
||
path = work_dir / f"gruppierung-{h}-c{tag}.json"
|
||
if _umbrella_schema(_json_file(path), all_ids) is None:
|
||
status, _v = await run_single_slot(
|
||
ctx, f"Gruppierung {tag}", key=f"blocks-{topic}-gruppierung-{h}-c{tag}",
|
||
prompt=_prompt("Blocks-Gruppierung", topic=topic, candidates=cand_text,
|
||
list=full_list, out_path=path),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=path: _umbrella_schema(_json_file(p), all_ids),
|
||
timeout=_timeout("research_mapping", count))
|
||
if status == CANCELLED:
|
||
return None
|
||
return path
|
||
|
||
# ONE wave: TOP + all cluster judges in parallel. The used-ties precedence lives in the
|
||
# ORDER of `sources` (TOP first), not in execution order — a failed judge simply leaves
|
||
# no valid file and is skipped in the collection below (legacy semantics).
|
||
jobs = [("TOP", "Scan the ENTIRE block list below and propose EVERY genuine umbrella "
|
||
"you find — do not restrict yourself to any subset.", n)]
|
||
jobs += [(str(ci), "\n".join(f"{g + 1}. {texts[g]}" for g in cluster), len(cluster))
|
||
for ci, cluster in enumerate(multi)]
|
||
await asyncio.gather(*[_assess(tag, cand, count) for tag, cand, count in jobs],
|
||
return_exceptions=True)
|
||
if ctx.is_cancelled():
|
||
return
|
||
sources = [work_dir / f"gruppierung-{h}-c{tag}.json" for tag, _, _ in jobs] # TOP first → wins used-ties
|
||
|
||
existing_norms = {r["title_norm"] for r in rows}
|
||
for other in await db.kanban_cards(topic, board=BOARD, kind="block"):
|
||
existing_norms.add(_norm_title(other["payload"].get("title", "")))
|
||
used: set[int] = set()
|
||
seen_norm = set(existing_norms)
|
||
chosen, skipped = [], []
|
||
for src in sources:
|
||
for title, desc, members in (_umbrella_schema(_json_file(src), all_ids) or []):
|
||
import re as _re
|
||
title = _re.sub(r"\s+[—–]\s+", ": ", title).strip()
|
||
members = [m for m in members if m not in used]
|
||
if len(members) < 2:
|
||
continue
|
||
mrows = [rows[m - 1] for m in members]
|
||
if any(_GROUP_STANDALONE.search(r["title"]) for r in mrows):
|
||
skipped.append({"umbrella": title, "grund": "type-gate",
|
||
"mitglieder": [r["title"] for r in mrows]})
|
||
continue
|
||
mc = _min_cos([m - 1 for m in members])
|
||
if mc < GROUP_MIN_COS_FLOOR:
|
||
skipped.append({"umbrella": title, "grund": "min-cos", "min_cos": mc,
|
||
"mitglieder": [r["title"] for r in mrows]})
|
||
continue
|
||
unorm = _norm_title(title)
|
||
member_norms = {r["title_norm"] for r in mrows}
|
||
if unorm not in member_norms and unorm in seen_norm:
|
||
skipped.append({"umbrella": title, "grund": "title-collision",
|
||
"mitglieder": [r["title"] for r in mrows]})
|
||
continue
|
||
used.update(members)
|
||
seen_norm.add(unorm)
|
||
chosen.append({"umbrella": title, "description": desc, "min_cos": mc, "members": members})
|
||
# reconcile: same parent proposed twice under different titles → union
|
||
if len(chosen) >= 2:
|
||
uv = await _vec_rows(flow, [f"{c['umbrella']} — {c['description']}" for c in chosen])
|
||
if uv is not None:
|
||
u_sims = uv @ uv.T
|
||
parent = list(range(len(chosen)))
|
||
for i in range(len(chosen)):
|
||
for j in range(i + 1, len(chosen)):
|
||
if float(u_sims[i][j]) >= GROUP_RECONCILE_FLOOR:
|
||
embedding._union(parent, i, j)
|
||
comp: dict[int, list[int]] = {}
|
||
for i in range(len(chosen)):
|
||
comp.setdefault(embedding._find(parent, i), []).append(i)
|
||
merged = []
|
||
for grp in comp.values():
|
||
if len(grp) == 1:
|
||
merged.append(chosen[grp[0]])
|
||
continue
|
||
rep = max(grp, key=lambda gi: len(chosen[gi]["members"]))
|
||
members = list(dict.fromkeys(m for gi in grp for m in chosen[gi]["members"]))
|
||
merged.append({**chosen[rep],
|
||
"description": " · ".join(chosen[gi]["description"] for gi in grp),
|
||
"members": members})
|
||
chosen = merged
|
||
# completion: absorb leftover standalone parts (per-member type-guard veto)
|
||
leftover = sorted(all_ids - used)
|
||
if chosen and leftover:
|
||
cp = work_dir / f"gruppierung-completion-{h}.json"
|
||
add = _completion_schema(_json_file(cp), len(chosen), set(leftover))
|
||
if add is None:
|
||
anchors = "\n".join(
|
||
f"UMBRELLA {k}: {c['umbrella']} — {c['description']}\n bereits: "
|
||
+ ", ".join(rows[m - 1]["title"] for m in c["members"]) for k, c in enumerate(chosen))
|
||
rest = "\n".join(f"{i}. {texts[i - 1]}" for i in leftover)
|
||
status, add = await run_single_slot(
|
||
ctx, "Gruppierung completion", key=f"blocks-{topic}-gruppierung-completion-{h}",
|
||
prompt=_prompt("Blocks-Gruppierung-Completion", topic=topic, umbrellas=anchors,
|
||
rest=rest, out_path=cp),
|
||
role="judge", capabilities="files",
|
||
payload=lambda result, p=cp: _completion_schema(_json_file(p), len(chosen), set(leftover)),
|
||
timeout=_timeout("research_mapping", len(leftover)))
|
||
if status == CANCELLED:
|
||
return
|
||
if status != OK:
|
||
add = []
|
||
for k, new_members in (add or []):
|
||
for m in new_members:
|
||
if m in used or not (1 <= m <= n) or _GROUP_STANDALONE.search(rows[m - 1]["title"]):
|
||
continue
|
||
used.add(m)
|
||
chosen[k]["members"].append(m)
|
||
# apply: umbrella card + member moves
|
||
moves = []
|
||
for c in chosen:
|
||
mrows = [rows[m - 1] for m in c["members"]]
|
||
readers = sorted(set().union(*[set(r["payload"].get("readers") or []) for r in mrows]))
|
||
srcs = sorted(set().union(*[set(r["payload"].get("sources") or []) for r in mrows]))
|
||
await db.kanban_upsert_card(topic, BOARD, f"b-u-{uuid.uuid4().hex[:8]}", "block", "gap_check", {
|
||
"title": clean_title(c["umbrella"]), "description": c["description"],
|
||
"readers": readers, "sources": srcs, "umbrella": True,
|
||
"children": [r["title"] for r in mrows],
|
||
})
|
||
for r in mrows:
|
||
r["payload"].update(reason="umbrella", merged_into=c["umbrella"])
|
||
await db.kanban_set_payload(topic, BOARD, r["card_id"], r["payload"])
|
||
moves.append((r["card_id"], "grouped"))
|
||
grouped = {cid for cid, _ in moves}
|
||
moves += [(r["card_id"], "gap_check") for r in rows if r["card_id"] not in grouped]
|
||
atomic_write_json(work_dir / "inventar-gruppierung.json",
|
||
{"vorher": n, "umbrellas": [{"umbrella": c["umbrella"],
|
||
"mitglieder": [rows[m - 1]["title"] for m in c["members"]]}
|
||
for c in chosen], "skipped": skipped}, indent=1)
|
||
if chosen:
|
||
_log(topic, f"Gruppierung: {len(chosen)} Umbrella(s), −{len(grouped)} Blöcke")
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_gap_check(ctx: GenContext, flow: Flow, cards):
|
||
"""BARRIER/drain — the finished blocks advance to `done` IMMEDIATELY (board 2 starts
|
||
while the supplement searches). ONE supplement round runs as an async PRODUCER: the
|
||
producer count keeps the flow alive and the research_done gates closed, so the new
|
||
titles feed back through every gate in a clean second pass. Failure never fatal."""
|
||
topic = flow.topic
|
||
if not flow.state.get("supplement_done"):
|
||
flow.state["supplement_done"] = True # one round (loop stop), set before the task
|
||
titles = [c["payload"].get("title", "") for c in cards]
|
||
titles += [b["title"] for b in await _context_blocks(topic, exclude=set())]
|
||
flow.add_producer() # SYNC before create_task — gates/exit must see the producer
|
||
|
||
async def _run():
|
||
try:
|
||
await _supplement_producer(ctx, flow, sorted({t for t in titles if t}))
|
||
except Exception:
|
||
log.exception("[%s] supplement failed", topic)
|
||
finally:
|
||
flow.done_producer()
|
||
|
||
asyncio.create_task(_run())
|
||
await db.kanban_advance_many(topic, BOARD, [(c["card_id"], "done") for c in cards])
|
||
flow.wake.set()
|
||
|
||
|
||
async def _supplement_beleg(ctx: GenContext, flow: Flow, supplements: list) -> list:
|
||
"""Evidence gate for supplement proposals: keyword excerpts per proposal, ONE no-tool
|
||
judge marks material coverage (ja/nein). Proposals without any matching excerpt drop
|
||
immediately; a failed gate keeps nothing (creep is costlier than a lost bonus round)."""
|
||
topic = flow.topic
|
||
folder = source_folder(topic)
|
||
packs = [(t, d, _evidence_pack(folder, None, [t], budget=6000)) for t, d in supplements]
|
||
cands = [(t, d, ev) for t, d, ev in packs if ev]
|
||
kept: list = []
|
||
if cands:
|
||
path = flow.work_dir / "supplement-beleg.json"
|
||
ids = set(range(1, len(cands) + 1))
|
||
verdict = _yesno_schema(_json_file(path), ids)
|
||
if verdict is None:
|
||
lines = "\n\n".join(f"{k}. {t} — {d}\nAUSZÜGE:\n{ev}"
|
||
for k, (t, d, ev) in enumerate(cands, 1))
|
||
status, verdict = await run_single_slot(
|
||
ctx, "Supplement-Beleg", key=f"blocks-{topic}-supplement-beleg",
|
||
prompt=_prompt("Blocks-Supplement-Beleg", topic=topic, proposals=lines,
|
||
extra=_extra(flow.state.get("instructions", ""))),
|
||
role="judge", capabilities="none",
|
||
payload=lambda result, p=path, i=ids: _sink_json(result, p, lambda d2: _yesno_schema(d2, i)),
|
||
timeout=_timeout("selection_mapping", len(cands)))
|
||
if status != OK or not isinstance(verdict, dict):
|
||
verdict = {}
|
||
kept = [(t, d) for k, (t, d, ev) in enumerate(cands, 1) if str(verdict.get(k, "nein")) == "ja"]
|
||
if len(kept) < len(supplements):
|
||
_log(topic, f"Supplement: {len(supplements) - len(kept)}/{len(supplements)} ohne Materialbeleg verworfen")
|
||
return kept
|
||
|
||
|
||
async def _supplement_producer(ctx: GenContext, flow: Flow, titles: list[str]):
|
||
"""One web agent proposes canonically missing blocks → new title cards (reader
|
||
'supplement' skips only the ≥2 consensus bar, every other gate applies)."""
|
||
topic = flow.topic
|
||
path = flow.work_dir / "supplement.json"
|
||
supplements = _supplement_schema(_json_file(path))
|
||
if supplements is None:
|
||
# Source-bound topics (uni/projekt/link): the MATERIAL defines the scope — the agent
|
||
# compares inventory vs. material (files, no web). Only pure "thema" topics research
|
||
# the canon on the web (measured: the web agent proposed 22 textbook blocks the
|
||
# script never treats, all discarded by the Beleg gate — 7 wasted minutes).
|
||
folder = source_folder(topic)
|
||
template, caps = ("Blocks-Supplement-Material", "files") if folder else ("Blocks-Supplement", "full")
|
||
status, supplements = await run_single_slot(
|
||
ctx, "Supplement", key=f"blocks-{topic}-supplement",
|
||
prompt=_prompt(template, topic=topic, project=folder,
|
||
blocks="\n".join(f"- {t}" for t in titles),
|
||
out_path=path, extra=_extra(flow.state.get("instructions", ""))),
|
||
role="quick", capabilities=caps,
|
||
payload=lambda result, p=path: _supplement_schema(_json_file(p)),
|
||
timeout=_timeout("ergaenzung"))
|
||
if status == CANCELLED:
|
||
flow.state["supplement_done"] = False # in-memory only; resume re-derives from the file
|
||
return
|
||
if status != OK:
|
||
_log(topic, "Supplement fehlgeschlagen — übersprungen (optional)")
|
||
supplements = []
|
||
# Material anchoring (uni/projekt): the web agent proposes canonical knowledge BLIND
|
||
# to the source (measured: 22/107 aak blocks were textbook standard the script never
|
||
# treats). Only proposals the material itself covers may enter the inventory.
|
||
if supplements and source_folder(topic):
|
||
supplements = await _supplement_beleg(ctx, flow, supplements)
|
||
# Dead lineage: blocks demoted by the fragment filter (and their cluster + title cards)
|
||
# must NOT dedup a supplement proposal — their content is gone. A hit on a dead title
|
||
# REOPENS the lineage instead: the title card rejoins its cluster (live re-cluster) and
|
||
# the respawned block gets a fresh fragment_filter pass. failed-quorum/pre-reject stay
|
||
# in the dedup: those were rejected as non-blocks, not lost as content.
|
||
dead_reasons = {"fragment", "drop-collateral", "drop"}
|
||
cards = await db.kanban_cards(topic, board=BOARD)
|
||
dead_clusters = {c["payload"].get("cluster") for c in cards
|
||
if c["kind"] == "block" and c["stage"] == "rejected"
|
||
and c["payload"].get("reason") in dead_reasons}
|
||
dead_clusters.discard(None)
|
||
membership = await db.kanban_membership(topic)
|
||
dead_titles = {nm for nm, cid in membership.items() if cid in dead_clusters}
|
||
|
||
def _is_dead(c) -> bool:
|
||
if c["kind"] == "title":
|
||
return c["card_id"] in dead_titles
|
||
if c["kind"] == "cluster":
|
||
return c["card_id"] in dead_clusters
|
||
return c["stage"] == "rejected" and c["payload"].get("reason") in dead_reasons
|
||
|
||
known_norms, known_keys = set(), set()
|
||
dead_by_norm: dict[str, str] = {} # title norm/key → requeue-able title card_id
|
||
dead_by_key: dict[str, str] = {}
|
||
for c in cards:
|
||
tt = c["payload"].get("title", "")
|
||
if not tt:
|
||
continue
|
||
if _is_dead(c):
|
||
if c["kind"] == "title":
|
||
dead_by_norm.setdefault(c["card_id"], c["card_id"])
|
||
if (k := _canonical_key(tt)):
|
||
dead_by_key.setdefault(k, c["card_id"])
|
||
continue
|
||
known_norms.add(_norm_title(tt))
|
||
if (k := _canonical_key(tt)):
|
||
known_keys.add(k)
|
||
new = reopened = 0
|
||
for t, d in (supplements or []):
|
||
t, d = clean_title(t), clean_title(d)
|
||
norm = _norm_title(t)
|
||
key = _canonical_key(t)
|
||
if not norm or norm in known_norms or (key and key in known_keys):
|
||
continue
|
||
known_norms.add(norm)
|
||
dead_id = dead_by_norm.get(norm) or (dead_by_key.get(key) if key else None)
|
||
if dead_id:
|
||
card = await db.kanban_get_card(topic, BOARD, dead_id)
|
||
if card:
|
||
card["payload"]["supplement"] = True
|
||
await db.kanban_set_payload(topic, BOARD, dead_id, card["payload"])
|
||
await db.kanban_advance(topic, BOARD, dead_id, "cluster")
|
||
reopened += 1
|
||
continue
|
||
desc = f"{d} [Supplement]".strip()
|
||
async with _ingest_lock:
|
||
await db.kanban_add_title(topic, BOARD, norm, t, desc, "supplement", "supplement")
|
||
card = await db.kanban_get_card(topic, BOARD, norm)
|
||
if card:
|
||
card["payload"]["supplement"] = True
|
||
await db.kanban_set_payload(topic, BOARD, norm, card["payload"])
|
||
new += 1
|
||
if new or reopened:
|
||
_log(topic, f"Supplement: {new} Block-Kandidat(en) → ingest, {reopened} wiedereröffnet")
|
||
flow.wake.set()
|
||
|
||
|
||
async def _proc_done(ctx: GenContext, flow: Flow, cards):
|
||
"""Mirror finished blocks into the legacy `blocks` table (status consensus) — the
|
||
interface everything downstream (guide, exam, frontend) already reads. Unique-title
|
||
collisions get a numeric suffix; a re-run under a new name discards the old mirror."""
|
||
topic = flow.topic
|
||
mirrored: dict[str, str] = flow.state.setdefault("mirrored", {})
|
||
moves = []
|
||
for c in cards:
|
||
p = c["payload"]
|
||
base = p.get("title", "")
|
||
if not base:
|
||
moves.append((c["card_id"], "done_block"))
|
||
continue
|
||
title, norm, i = base, _norm_title(base), 2
|
||
while mirrored.get(norm) not in (None, c["card_id"]):
|
||
title = f"{base} ({i})"
|
||
norm = _norm_title(title)
|
||
i += 1
|
||
old = p.get("mirrored_norm")
|
||
if old and old != norm:
|
||
await db.set_block_status(topic, old, "discarded")
|
||
await db.upsert_block(topic, norm, title, p.get("description", ""), p.get("sources") or [])
|
||
await db.set_block_status(topic, norm, "consensus",
|
||
title=title, description=p.get("description", ""))
|
||
mirrored[norm] = c["card_id"]
|
||
p.update(mirrored_norm=norm, title=title)
|
||
await db.kanban_set_payload(topic, BOARD, c["card_id"], p)
|
||
if (spawn := flow.state.get("spawn_artefact")):
|
||
await spawn(c["card_id"], p) # board 2 card per finished block
|
||
moves.append((c["card_id"], "done_block"))
|
||
await db.kanban_advance_many(topic, BOARD, moves)
|
||
flow.wake.set()
|
||
|
||
|
||
# ── Orchestration ──────────────────────────────────────────────────────────────────
|
||
def inventory_stages(ctx: GenContext, flow: Flow) -> list[Stage]:
|
||
research_done = lambda: flow.research_done # noqa: E731
|
||
return [
|
||
Stage(BOARD, "ingest", lambda cs: _proc_ingest(flow, cs)),
|
||
Stage(BOARD, "cluster", lambda cs: _proc_cluster(ctx, flow, cs), serial=True),
|
||
Stage(BOARD, "pair_check", lambda cs: _proc_pair_check(ctx, flow, cs)),
|
||
# gate (no barrier): hold clusters until ALL research producers are done — a title's
|
||
# second reader may arrive minutes later, and the ≥2 vote must count it.
|
||
Stage(BOARD, "consensus_gate", lambda cs: _proc_consensus_gate(ctx, flow, cs),
|
||
gate=research_done),
|
||
Stage(BOARD, "clarify", lambda cs: _proc_clarify(ctx, flow, cs)),
|
||
Stage(BOARD, "naming", lambda cs: _proc_naming(ctx, flow, cs)),
|
||
Stage(BOARD, "naming_check", lambda cs: _proc_naming_check(ctx, flow, cs)),
|
||
Stage(BOARD, "fragment_filter", lambda cs: _proc_fragment_filter(ctx, flow, cs),
|
||
barrier=True, drain=True, gate=research_done),
|
||
Stage(BOARD, "dedup", lambda cs: _proc_dedup(ctx, flow, cs),
|
||
barrier=True, drain=True, gate=research_done),
|
||
Stage(BOARD, "grouping", lambda cs: _proc_grouping(ctx, flow, cs),
|
||
barrier=True, drain=True, gate=research_done),
|
||
Stage(BOARD, "gap_check", lambda cs: _proc_gap_check(ctx, flow, cs),
|
||
barrier=True, drain=True, gate=research_done),
|
||
Stage(BOARD, "done", lambda cs: _proc_done(ctx, flow, cs)),
|
||
]
|
||
|
||
|
||
async def _preload_state(flow: Flow):
|
||
"""Continue/resume: rebuild the in-memory caches from the persisted cards."""
|
||
titles = await db.kanban_cards(flow.topic, board=BOARD, kind="title")
|
||
known = [t for t in titles if t["stage"] != "ingest"]
|
||
flow.state["nn"] = {"norms": [t["card_id"] for t in known],
|
||
"texts": [_t_text(t["payload"]) for t in known]}
|
||
flow.state["mirrored"] = {
|
||
c["payload"]["mirrored_norm"]: c["card_id"]
|
||
for c in await db.kanban_cards(flow.topic, board=BOARD, stage="done_block")
|
||
if c["payload"].get("mirrored_norm")}
|
||
flow.state["supplement_done"] = _supplement_schema(
|
||
_json_file(flow.work_dir / "supplement.json")) is not None
|
||
|
||
|
||
async def run_boards(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str,
|
||
research: bool = True, artefacts: bool = True, qa_force: bool = False) -> bool:
|
||
"""Run the inventory board (plus board 2 „Artefakte") until quiescence.
|
||
research=False = Continue: drain the existing queue, search nothing new.
|
||
qa_force=True overrides a failed QA gate (user clicked „Trotzdem fortsetzen")."""
|
||
topic = ctx.topic
|
||
flow = Flow(topic, files["arbeit"])
|
||
flow.state["instructions"] = instructions
|
||
flow.state["qa_force"] = qa_force
|
||
run_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M')}-{uuid.uuid4().hex[:4]}"
|
||
flow.state["run_id"] = run_id
|
||
flow.state["run_started"] = datetime.now(timezone.utc).isoformat()
|
||
db.set_current_run(topic, run_id) # every event of this flow carries the run_id
|
||
if q["type"] == "link" and folder:
|
||
pages = await db.list_content(topic)
|
||
flow.state["pages"] = pages or sorted(set(_crawl_index(folder).values()))
|
||
await _preload_state(flow)
|
||
stages = inventory_stages(ctx, flow)
|
||
inv_names = [st.stage for st in stages]
|
||
if artefacts:
|
||
import board_artefacts # lazy — board_artefacts imports blocks too
|
||
flow.state["spawn_artefact"] = board_artefacts.make_spawner(topic, files)
|
||
await board_artefacts.ensure_outline_card(topic)
|
||
stages += board_artefacts.artefact_stages(ctx, flow, files, q, folder, instructions)
|
||
if QA_GATE_NOTE > 0:
|
||
# QA gate: board 2 waits until the inventory QA passed (or the user forces).
|
||
# Costs pipelining (board 2 no longer starts per finished block) but saves
|
||
# tokens on a bad foundation — the watcher below runs the QA and decides.
|
||
sub = next(st for st in stages if st.stage == "subblocks")
|
||
sub.gate = lambda: bool(flow.state.get("qa_ok") or flow.state.get("qa_force"))
|
||
stages = chain_stages(stages)
|
||
if artefacts:
|
||
# Outline needs every block's TITLE + FACTS, nothing later: cut the post-facts
|
||
# artefact stages from its barrier so it runs parallel to levels…finalize of the
|
||
# slowest block (makespan tail). Inventory stages all stay — no late blocks.
|
||
outline = next(s for s in stages if s.stage == "outline")
|
||
outline.upstream = [u for u in outline.upstream if u not in
|
||
("levels", "relevance", "question_pattern", "artefacts", "finalize")]
|
||
producers = _build_producers(ctx, flow, q, folder, instructions) if research else []
|
||
|
||
async def _as_producer(coro):
|
||
try:
|
||
await coro
|
||
except Exception:
|
||
log.exception("[%s] research producer failed", topic)
|
||
finally:
|
||
flow.done_producer()
|
||
|
||
for _ in producers:
|
||
flow.add_producer()
|
||
flow.spawn_research = lambda: _research_once(
|
||
ctx, flow, q, folder, instructions, f"x{flow.next_tag()}")
|
||
# cancel hook: blocks.cancel_blocks flips is_cancelled; stop the flow with it
|
||
stopper = asyncio.create_task(_stop_on_cancel(ctx, flow))
|
||
watcher = (asyncio.create_task(_qa_gate_watch(ctx, flow, inv_names, set_p))
|
||
if artefacts and QA_GATE_NOTE > 0 else None)
|
||
try:
|
||
await kanban.run_flow(flow, stages, [_as_producer(p) for p in producers], set_p)
|
||
finally:
|
||
stopper.cancel()
|
||
if watcher:
|
||
watcher.cancel()
|
||
db.set_current_run(topic, None)
|
||
if ctx.is_cancelled():
|
||
return False
|
||
if flow.state.get("qa_paused"):
|
||
return True # kein Fehler: Flow hielt am QA-Gate, Karten warten in subblocks
|
||
await _write_final(topic, files)
|
||
await _write_run_summary(topic, flow)
|
||
return True
|
||
|
||
|
||
_QA_GATE_POLL = 2.0 # Sekunden zwischen Quiescence-Checks des QA-Wächters
|
||
|
||
|
||
async def _qa_gate_watch(ctx: GenContext, flow: Flow, inv_names: list[str], set_p):
|
||
"""Companion task: once the inventory is quiescent, run the QA once and decide —
|
||
open the board-2 gate or pause the flow. Fail-OPEN on errors (QA is a helper,
|
||
not a jailer); qa_force short-circuits to open."""
|
||
topic = flow.topic
|
||
try:
|
||
while not flow.stop:
|
||
if (flow.research_done and not flow.active_in(inv_names)
|
||
and await db.kanban_count(topic, inv_names) == 0):
|
||
break
|
||
await asyncio.sleep(_QA_GATE_POLL)
|
||
if flow.stop or ctx.is_cancelled():
|
||
return
|
||
if flow.state.get("qa_force"):
|
||
flow.state["qa_ok"] = True
|
||
flow.wake.set()
|
||
return
|
||
set_p("QA prüft das Inventar…")
|
||
import qa
|
||
report = await qa.qa_report(topic, llm=QA_GATE_LLM)
|
||
note = float(report["note"]) if report else 10.0
|
||
flow.state["qa_note"] = note
|
||
if report:
|
||
try: # Report-Persistenz ist Komfort — ein Schreibfehler darf das Gate nicht öffnen
|
||
await asyncio.to_thread(qa._write_report, report)
|
||
except Exception:
|
||
log.exception("[%s] QA-Report schreiben fehlgeschlagen", topic)
|
||
if note >= QA_GATE_NOTE:
|
||
_log(topic, f"QA-Gate: Note {note} ≥ {QA_GATE_NOTE} — Board 2 startet")
|
||
flow.state["qa_ok"] = True
|
||
else:
|
||
_log(topic, f"QA-Gate: Note {note} < {QA_GATE_NOTE} — pausiert (Continue erzwingt)")
|
||
set_p(f"QA-Note {note} < {QA_GATE_NOTE} — pausiert")
|
||
flow.state["qa_paused"] = True
|
||
flow.stop = True
|
||
flow.wake.set()
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
log.exception("[%s] QA-Gate fehlgeschlagen — Gate offen (fail-open)", topic)
|
||
flow.state["qa_ok"] = True
|
||
flow.wake.set()
|
||
|
||
|
||
async def _write_run_summary(topic: str, flow: Flow):
|
||
"""lauf-summary.json: the per-run numbers block QA diffs against. Never fatal."""
|
||
try:
|
||
run_id = flow.state.get("run_id", "")
|
||
started = flow.state.get("run_started", "")
|
||
finished = datetime.now(timezone.utc).isoformat()
|
||
dauer = ""
|
||
if started:
|
||
dauer = round((datetime.fromisoformat(finished) - datetime.fromisoformat(started)).total_seconds() / 60, 1)
|
||
summary = {"run_id": run_id, "topic": topic, "started": started, "finished": finished,
|
||
"dauer_min": dauer, "boards": await db.kanban_stage_counts(topic),
|
||
**await db.events_run_summary(topic, run_id)}
|
||
try: # Abschluss-QA MIT Judges: sub_dubletten/unechte werden beurteilt — erst damit
|
||
import qa # ist note_artefakte belastbar (Kandidatenliste allein zählt nicht)
|
||
report = await qa.qa_report(topic, llm=True)
|
||
if report:
|
||
summary["note"] = report["note"]
|
||
summary["note_artefakte"] = report.get("note_artefakte")
|
||
summary["artefakte"] = report.get("artefakte", {})
|
||
await asyncio.to_thread(qa._write_report, report)
|
||
except Exception:
|
||
log.exception("[%s] Abschluss-QA fehlgeschlagen", topic)
|
||
atomic_write_json(flow.work_dir / "lauf-summary.json", summary, indent=1)
|
||
except Exception:
|
||
log.exception("[%s] lauf-summary fehlgeschlagen", topic)
|
||
|
||
|
||
async def _stop_on_cancel(ctx: GenContext, flow: Flow):
|
||
while not flow.stop:
|
||
if ctx.is_cancelled():
|
||
flow.stop = True
|
||
flow.wake.set()
|
||
return
|
||
await asyncio.sleep(0.3)
|
||
|
||
|
||
async def _write_final(topic: str, files: dict):
|
||
"""blocks.md from the finished cards (flow order — titles are already unique)."""
|
||
done = await db.kanban_cards(topic, board=BOARD, stage="done_block")
|
||
done.sort(key=lambda c: c["updated_at"])
|
||
lines = [_line(i, c["payload"]) for i, c in enumerate(done, 1) if c["payload"].get("title")]
|
||
if lines:
|
||
atomic_write_text(files["final"], "\n".join(lines) + "\n")
|
||
_log(topic, f"Kanban: fertig — {len(lines)} Blöcke")
|
||
|
||
|
||
# ── Board API (snapshot, reset, dead-letter) ───────────────────────────────────────
|
||
# (board, stage, label, kind) — display order of the live board. Terminal columns last.
|
||
COLUMNS = [
|
||
("inventory", "ingest", "Eingang", "title"),
|
||
("inventory", "cluster", "Cluster", "title"),
|
||
("inventory", "pair_check", "Paar-Check", "cluster"),
|
||
("inventory", "consensus_gate", "Konsens", "cluster"),
|
||
("inventory", "clarify", "Klärung", "cluster"),
|
||
("inventory", "naming", "Naming", "cluster"),
|
||
("inventory", "naming_check", "Naming-Check", "cluster"),
|
||
("inventory", "fragment_filter", "Fragment-Filter", "block"),
|
||
("inventory", "dedup", "Dubletten", "block"),
|
||
("inventory", "grouping", "Gruppierung", "block"),
|
||
("inventory", "gap_check", "Lücken-Check", "block"),
|
||
("inventory", "done", "Spiegeln", "block"),
|
||
("inventory", "done_block", "Fertig", "block"),
|
||
("inventory", "rejected", "Verworfen", None),
|
||
("inventory", "grouped", "Zusammengelegt", "block"),
|
||
("artefacts", "subblocks", "Subbausteine", "ablock"),
|
||
("artefacts", "facts", "Fakten", "ablock"),
|
||
("artefacts", "levels", "Stufen", "ablock"),
|
||
("artefacts", "relevance", "Relevanz", "ablock"),
|
||
("artefacts", "konsolidierung", "Konsolidierung", "ablock"),
|
||
("artefacts", "question_pattern", "Fragen", "ablock"),
|
||
("artefacts", "artefacts", "Lernkarten", "ablock"),
|
||
("artefacts", "finalize", "Zusammenführen", "ablock"),
|
||
("artefacts", "outline", "Gliederung", "outline"),
|
||
("artefacts", "done_artefact", "Fertig", "ablock"),
|
||
]
|
||
|
||
_TITLE_STAGES = ["ingest", "cluster"]
|
||
_CLUSTER_STAGES = ["pair_check", "consensus_gate", "clarify", "naming", "naming_check"]
|
||
_BLOCK_STAGES = ["fragment_filter", "dedup", "grouping", "gap_check", "done"]
|
||
_ART_STAGES = ["subblocks", "facts", "levels", "relevance", "konsolidierung", "question_pattern",
|
||
"artefacts", "finalize", "outline"]
|
||
# where a requeued dead card restarts, by kind
|
||
_DEAD_RESTART = {"title": "cluster", "cluster": "pair_check", "block": "fragment_filter",
|
||
"ablock": "subblocks", "outline": "outline"}
|
||
_VERDICT_KEYS = ("reason", "votes", "judges", "merged_into", "parent_norm", "mirrored_norm")
|
||
DONE_ART = "done_artefact"
|
||
|
||
|
||
# Board-2 stage → its fine-step group in blocks.PHASEN (drives the per-card stepper).
|
||
_STAGE_PHASE = {"subblocks": "Subblocks", "facts": "Facts", "levels": "Levels",
|
||
"relevance": "Relevance", "question_pattern": "Questions", "artefacts": "Artefacts"}
|
||
_PHASE_STEPS = {name: steps for name, steps in blocks.PHASEN}
|
||
|
||
|
||
def _card_view(r: dict, active: set[str], live_info: dict) -> dict:
|
||
p = r["payload"]
|
||
key = f"{r['board']}:{r['card_id']}"
|
||
is_active = key in active
|
||
info = r.get("last_error") or p.get("reason") or ""
|
||
if p.get("merged_into"):
|
||
info = f"→ {p['merged_into']}"
|
||
elif p.get("parent_norm"):
|
||
info = f"Fragment von: {p['parent_norm']}"
|
||
out = {"title": p.get("title") or r["card_id"], "retries": r["retries"], "card_id": r["card_id"],
|
||
"kind": r.get("kind", ""), "board": r.get("board", ""),
|
||
"status": "error" if r["retries"] else ("active" if is_active else "open")}
|
||
live = live_info.get(key)
|
||
if is_active and live: # live step message wins while the card is worked
|
||
info = live["msg"] if isinstance(live, dict) else live # dict since the stepper, str before
|
||
step = live.get("step") if isinstance(live, dict) else ""
|
||
steps = _PHASE_STEPS.get(_STAGE_PHASE.get(r.get("stage", ""), ""), ())
|
||
if step in steps: # phase stepper: which fine step of the card's stage runs (1-based)
|
||
out.update(step_i=steps.index(step) + 1, step_n=len(steps), steps=list(steps))
|
||
out["info"] = info
|
||
return out
|
||
|
||
|
||
async def board_snapshot(topic: str, limit: int = 20) -> dict:
|
||
"""Live board: per column count + the newest cards (title, status, info)."""
|
||
counts = await db.kanban_stage_counts(topic)
|
||
flow = kanban.active_flows.get(topic)
|
||
active = flow.active_cards if flow else set()
|
||
live_info = flow.state.get("card_info", {}) if flow else {}
|
||
columns = []
|
||
for board, stage, label, _kind in COLUMNS:
|
||
total = counts.get(board, {}).get(stage, 0)
|
||
cards = ([_card_view(r, active, live_info)
|
||
for r in await db.kanban_stage_cards(topic, board, stage, limit)]
|
||
if total else [])
|
||
columns.append({"board": board, "key": stage, "label": label, "total": total, "cards": cards})
|
||
dead = [{"card_id": r["card_id"], "board": r["board"], "kind": r["kind"],
|
||
"title": r["payload"].get("title") or r["card_id"], "error": r.get("last_error") or ""}
|
||
for r in await db.kanban_dead(topic)]
|
||
return {"columns": columns, "dead": dead,
|
||
"done": counts.get("inventory", {}).get("done_block", 0),
|
||
"qa": _qa_view(topic, counts, flow)}
|
||
|
||
|
||
def _qa_view(topic: str, counts: dict, flow) -> dict | None:
|
||
"""Latest QA report digest for the board header. `pausiert` = the gate stopped the
|
||
flow (score below threshold, board-2 cards waiting, no flow running)."""
|
||
# no inventory (deleted/never built) → no badge; the report files stay on purpose,
|
||
# so the first run after a rebuild diffs against the old state
|
||
if not counts.get("inventory", {}).get("done_block", 0) and not (
|
||
flow and flow.state.get("qa_note") is not None):
|
||
return None
|
||
import qa
|
||
tdir = qa.QA_DIR / topic
|
||
# by mtime: a re-run overwrites the run-id-named file, which sorts before timestamp names.
|
||
# guide-* reports are the guide_qa series — they must not shadow the inventory badge.
|
||
reports = sorted((p for p in tdir.glob("*.json") if not p.name.startswith("guide-")),
|
||
key=lambda p: p.stat().st_mtime) if tdir.is_dir() else []
|
||
if not reports:
|
||
return None
|
||
r = _json_file(reports[-1]) or {}
|
||
note = r.get("note")
|
||
if note is None:
|
||
return None
|
||
wartend = counts.get("artefacts", {}).get("subblocks", 0)
|
||
pausiert = bool(note < QA_GATE_NOTE and wartend and flow is None)
|
||
return {"note": note, "note_artefakte": r.get("note_artefakte"),
|
||
"schwelle": QA_GATE_NOTE, "pausiert": pausiert,
|
||
"quoten": r.get("quoten", {}),
|
||
"befunde": (r.get("fremd", []) + r.get("unecht", []))[:6]}
|
||
|
||
|
||
async def _clean_artefact_state(topic: str, files: dict) -> None:
|
||
"""Artefact data is fully derived — wipe cards, DB tables and sidecar files."""
|
||
await db.kanban_delete_cards(topic, "artefacts")
|
||
await db.delete_subblocks(topic)
|
||
await db.delete_question_pattern(topic)
|
||
await db.delete_sub_artefakte(topic)
|
||
await db.delete_outline(topic)
|
||
for key in ("sub_roh", "sidecar", "facts", "question_pattern", "artefakte", "outline"):
|
||
files[key].unlink(missing_ok=True)
|
||
|
||
|
||
async def reset_board_from_stage(topic: str, board: str, stage: str, files: dict) -> int:
|
||
"""Reset ab Spalte: cards from `stage` onward (incl. terminal rejected/grouped/dead)
|
||
return to `stage`; downstream derived state is wiped. → moved card count.
|
||
Only call while nothing is generating (the route guards)."""
|
||
moved = 0
|
||
|
||
async def _requeue(card: dict, to_stage: str):
|
||
nonlocal moved
|
||
p = {k: v for k, v in card["payload"].items() if k not in _VERDICT_KEYS}
|
||
await db.kanban_set_payload(topic, card["board"], card["card_id"], p)
|
||
await db.kanban_advance(topic, card["board"], card["card_id"], to_stage)
|
||
moved += 1
|
||
|
||
if board == "inventory" and stage in _TITLE_STAGES:
|
||
# everything below titles is derived → full re-derive
|
||
for r in await db.kanban_cards(topic, board="inventory"):
|
||
if r["kind"] == "title":
|
||
await _requeue(r, stage)
|
||
await db.kanban_delete_cards(topic, "inventory", "cluster")
|
||
await db.kanban_delete_cards(topic, "inventory", "block")
|
||
dbc = await db.get_db()
|
||
await dbc.execute("DELETE FROM kanban_members WHERE topic = ?", (topic,))
|
||
await dbc.commit()
|
||
await db.delete_blocks(topic)
|
||
await _clean_artefact_state(topic, files)
|
||
elif board == "inventory" and stage in _CLUSTER_STAGES:
|
||
idx = _CLUSTER_STAGES.index(stage)
|
||
later = set(_CLUSTER_STAGES[idx:]) | {"done_cluster", "rejected", "dead"}
|
||
for r in await db.kanban_cards(topic, board="inventory", kind="cluster"):
|
||
if r["stage"] in later:
|
||
await _requeue(r, stage)
|
||
await db.kanban_delete_cards(topic, "inventory", "block")
|
||
await db.delete_blocks(topic)
|
||
await _clean_artefact_state(topic, files)
|
||
elif board == "inventory":
|
||
idx = _BLOCK_STAGES.index(stage) if stage in _BLOCK_STAGES else 0
|
||
later = set(_BLOCK_STAGES[idx:]) | {"done_block", "rejected", "grouped", "dead"}
|
||
for r in await db.kanban_cards(topic, board="inventory", kind="block"):
|
||
if r["stage"] in later:
|
||
await _requeue(r, stage)
|
||
await db.delete_blocks(topic)
|
||
await _clean_artefact_state(topic, files)
|
||
else: # artefacts board
|
||
idx = _ART_STAGES.index(stage) if stage in _ART_STAGES else 0
|
||
later = set(_ART_STAGES[idx:]) | {DONE_ART, "dead"}
|
||
for r in await db.kanban_cards(topic, board="artefacts"):
|
||
if r["kind"] == "outline":
|
||
await _requeue(r, "outline")
|
||
elif r["stage"] in later:
|
||
await _requeue(r, stage)
|
||
if stage == "subblocks": # full artefact re-derive → also the DB mirrors
|
||
await db.delete_subblocks(topic)
|
||
await db.delete_question_pattern(topic)
|
||
await db.delete_sub_artefakte(topic)
|
||
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
|