From c05421a8c18ade86a3226d946e2bf599429e8045 Mon Sep 17 00:00:00 2001 From: Team3 Date: Sat, 4 Jul 2026 19:20:48 +0200 Subject: [PATCH] update --- backend/blocks.py | 109 ++++++++++++++---- backend/board_artefacts.py | 152 ++++++++++++-------------- backend/board_inventory.py | 30 ++--- backend/config.py | 5 + backend/guide_board.py | 130 +++++++++++----------- backend/pipeline.py | 87 +++++++++++++-- backend/qa.py | 17 ++- backend/repair.py | 49 +++++---- backend/routes.py | 2 +- backend/tests/test_board_inventory.py | 32 ++++++ backend/tests/test_konsolidierung.py | 128 +++++++++++----------- backend/tests/test_qa.py | 21 ++++ backend/tests/test_race.py | 138 +++++++++++++++++++++++ backend/tests/test_subblocks.py | 72 +++++++++++- 14 files changed, 695 insertions(+), 277 deletions(-) create mode 100644 backend/tests/test_race.py diff --git a/backend/blocks.py b/backend/blocks.py index e39eba6..96aa5a1 100644 --- a/backend/blocks.py +++ b/backend/blocks.py @@ -31,7 +31,7 @@ from jsonio import parse_json_text, read_json_file as _json_file from paths import arbeit_dir, blocks_path, question_pattern_path, project_dir, subblocks_path, source_path, source_crawl_dir, safe_folder from crawl import crawl from pipeline import ( - CANCELLED, FAILED, OK, GenContext, _extra, _gather_progress, _yesno_schema, _log, _prompt, _race, + CANCELLED, FAILED, OK, GenContext, _detached, _extra, _gather_progress, _yesno_schema, _log, _prompt, _race, _relevance_schema, _runde_schema, _semaphore, _str_list, _levels_schema, _timeout, run_single_slot, ) from textkit import ( @@ -597,6 +597,39 @@ def _sink_json(result, path: Path, schema): return val +async def _panel_2of3(tasks: dict, sink, outs_now, norm) -> None: + """Panel-Welle „first 2 agree": kehrt zurück, sobald zwei vorliegende Verdicts + übereinstimmen — die dritte Stimme kann die Mehrheit dann nicht mehr kippen. Sonst + (Dissens) wird weiter gewartet. Der Langsamste bestimmte jede Welle (gemessen: 98 s + bei ok-p50 ~50 s). Nachzügler laufen detached weiter; ihr File dient nur dem Resume. + tasks: {Task: judge_nr} · sink(j, result) persistiert · outs_now() liest Verdicts · + norm(verdict) macht sie vergleichbar.""" + offen = dict(tasks) + while offen: + done, _rest = await asyncio.wait(list(offen), return_when=asyncio.FIRST_COMPLETED) + for t in done: + j = offen.pop(t) + try: + r = t.result() + except Exception: # noqa: BLE001 — Panel ist fail-open, Ausfall = fehlende Stimme + continue + if isinstance(r, tuple): + sink(j, r) + outs = [norm(s) for s in outs_now()] + if len(outs) >= 2 and any(outs[a] == outs[b] + for a in range(len(outs)) for b in range(a + 1, len(outs))): + break + for t, j in offen.items(): # Dritter läuft weiter — sein File zählt fürs Resume + async def _warte(t=t, j=j): + try: + r = await t + if isinstance(r, tuple): + sink(j, r) + except (asyncio.CancelledError, Exception): # noqa: BLE001 + pass + _detached(asyncio.create_task(_warte())) + + def _sink_subs(result, path: Path): """Finder reply as TEXT (marker format), persisted to `path` for audit/diagnosis. File fallback: a tool-capable agent (thema web mode) that wrote the file despite the @@ -796,7 +829,23 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i "role": "quick", "capabilities": round_caps, "payload": (lambda result, p=p: _sink_subs(result, p)), } for k, p in zip(keys, paths)] - agent_texts = await _race(topic, label, slots, 2, _timeout("subblock", len(subset)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) + + async def _fold_late(d: dict) -> None: + """Dritte Stimme nachbuchen statt warten (ersetzte grace=300, gemessen 73 s/Runde): + Mentions sind additiv; ein Fund, den nur der Nachzügler hat, bleibt Einzelfund + und läuft durchs Clarify-Quellen-Gate — verfälscht wird nichts.""" + for marker, subs in d.items(): + num = _resolve_title(chunk_idx, marker) + if num is None: + continue + seen_late = set() + for sub in subs: + sn = _norm_title(sub) + if sn and sn not in seen_late: + seen_late.add(sn) + await db.upsert_subblock(topic, norm_by_num[num], sn, title_by_num[num], sub) + + agent_texts = await _race(topic, label, slots, 2, _timeout("subblock", len(subset)), provider, cancelled=is_cancelled, late=_fold_late) if is_cancelled() or not agent_texts: return None rows_before = {num: await db.list_subblocks(topic, norm_by_num[num]) for num in subset} @@ -1846,16 +1895,29 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst fk[f] = ek[f] return raw + def _inline_source(queries: list[str]) -> tuple[str, str]: + """Korpus-Auszüge INLINE statt Datei-Recherche → (source, capabilities). Die + Tool-Agenten (bash/read) verloren sich messbar in Reasoning-Schleifen (bis 39k + Zeichen) und endeten mit leerem Turn — Retry-Wellen à 60–90 s seriell pro Block. + No-Tool-Calls mit Inline-Material hatten 0 solcher Fälle (Muster: Facts-Check). + Fail-open: ohne Treffer bleibt die alte Selbst-Recherche.""" + ev = _evidence_pack(folder, sources, queries) if folder else "" + if ev: + return _prompt("Blocks-Source-Inline", excerpts=ev), "none" + return source, caps + # Phase "Facts find": 1 generator per chunk. async def _find(ci, idxs): fp = raw_path(ci) if _facts_schema(_json_file(fp)): return True subs_total = sum(len(blocks[i][1]) for i in idxs) + f_source, f_caps = await asyncio.to_thread( + _inline_source, [blocks[i][0] for i in idxs] + [s for i in idxs for s in blocks[i][1]]) status, _r = await run_single_slot( ctx, f"{lbl}Facts {ci}", key=f"blocks-{topic}-{ns}facts-c{ci}", - prompt=_prompt("Facts-Research", topic=topic, source=source, blocks=block_text(idxs), out_path=fp, extra=_extra(instructions)), - role="quick", capabilities=caps, + prompt=_prompt("Facts-Research", topic=topic, source=f_source, blocks=block_text(idxs), out_path=fp, extra=_extra(instructions)), + role="quick", capabilities=f_caps, payload=lambda result, p=fp: _sink_or_file(result, p, _facts_schema), timeout=_timeout("content", subs_total)) return status != FAILED and _facts_schema(_json_file(fp)) is not None @@ -1883,10 +1945,12 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst for fk in fm.values()) for bt, fm in per.items()) subs_total = sum(len(blocks[i][1]) for i in idxs) + e_source, e_caps = await asyncio.to_thread( + _inline_source, [blocks[i][0] for i in idxs] + [s for i in idxs for s in blocks[i][1]]) await run_single_slot( ctx, f"{lbl}Facts supplement {ci}", key=f"blocks-{topic}-{ns}facts-erg-c{ci}", - prompt=_prompt("Facts-Supplement", topic=topic, source=source, blocks=block, out_path=ep, extra=_extra(instructions)), - role="quick", capabilities=caps, + prompt=_prompt("Facts-Supplement", topic=topic, source=e_source, blocks=block, out_path=ep, extra=_extra(instructions)), + role="quick", capabilities=e_caps, payload=lambda result, p=ep: _sink_or_file(result, p, _facts_schema), timeout=_timeout("content", subs_total)) @@ -1913,16 +1977,17 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst ev = _cited_evidence(folder, sources, cites, fallback) if folder else "" c_source = _prompt("Blocks-Source-Inline", excerpts=ev) if ev else source pending = [j for j in panel if _facts_check_schema(_json_file(chk_path(ci, j))) is None] - rs = await asyncio.gather(*[ + tmap = {asyncio.create_task( run_agent(f"blocks-{topic}-{ns}facts-check-c{ci}-j{j}", _prompt("Facts-Check", topic=topic, source=c_source, facts=facts_text, out_path=chk_path(ci, j), extra=_extra(instructions)), _timeout("content_check", len(per)), provider=provider, role="judge", capabilities="none" if ev else caps, - scope=topic, label=f"{lbl}Facts check {ci}/{j}") - for j in pending], return_exceptions=True) - for j, r in zip(pending, rs): - if isinstance(r, tuple): - _sink_json(r, chk_path(ci, j), _facts_check_schema) + scope=topic, label=f"{lbl}Facts check {ci}/{j}")): j + for j in pending} + await _panel_2of3(tmap, lambda j, r: _sink_json(r, chk_path(ci, j), _facts_check_schema), + lambda: [s for j in panel + if (s := _facts_check_schema(_json_file(chk_path(ci, j)))) is not None], + lambda s: {tuple(x) for x in s}) outs = [s for j in panel if (s := _facts_check_schema(_json_file(chk_path(ci, j)))) is not None] bvotes: dict[str, int] = {} vvotes: dict[str, int] = {} @@ -1970,10 +2035,13 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst goal.append(f"BLOCK: {bt}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in affected_subs)) if not goal: return + x_source, x_caps = await asyncio.to_thread( + _inline_source, [bt for bt in rel_by] + [s for subs in rel_by.values() + for s in subs if _norm_title(s) in subs_norm]) await run_single_slot( ctx, f"{lbl}Facts-Fix {ci}", key=f"blocks-{topic}-{ns}facts-fix-c{ci}", - prompt=_prompt("Facts-Research", topic=topic, source=source, blocks="\n\n".join(goal), out_path=fix_path(ci), extra=_extra(instructions)), - role="quick", capabilities=caps, + prompt=_prompt("Facts-Research", topic=topic, source=x_source, blocks="\n\n".join(goal), out_path=fix_path(ci), extra=_extra(instructions)), + role="quick", capabilities=x_caps, payload=lambda result, p=fix_path(ci): _sink_or_file(result, p, _facts_schema), timeout=_timeout("content", len(subs_norm))) await _gather_progress([_fix(ci) for ci in correctable], len(correctable), _report_p(set_p, topic, "Facts fix")) @@ -3266,15 +3334,16 @@ async def _artefacts_block(ctx: GenContext, set_p, files: dict, sidecar: dict, i pending = [j for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if _example_check_schema(_json_file(cpath(j))) is None] if pending: # ground truth (facts) is fully inline → no tools, text reply, engine persists - rs = await asyncio.gather(*[ + tmap = {asyncio.create_task( run_agent(f"blocks-{topic}-{ns}artifact-example-check-c{ci}-j{j}", _prompt("Artifact-Example-Check", topic=topic, facts=block_text(idxs), examples=examples_txt, out_path=cpath(j), extra=_extra(instructions)), _timeout("content_check", len(items)), provider=provider, role="judge", capabilities="none", - scope=topic, label=f"{lbl}Beispiel-Check {ci}/{j}") - for j in pending], return_exceptions=True) - for j, r in zip(pending, rs): - if isinstance(r, tuple): - _sink_json(r, cpath(j), _example_check_schema) + scope=topic, label=f"{lbl}Beispiel-Check {ci}/{j}")): j + for j in pending} + await _panel_2of3(tmap, lambda j, r: _sink_json(r, cpath(j), _example_check_schema), + lambda: [s for j in (1, 2, 3)[:FACTS_CHECK_PANEL] + if (s := _example_check_schema(_json_file(cpath(j)))) is not None], + lambda s: frozenset(s)) outs = [s for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if (s := _example_check_schema(_json_file(cpath(j)))) is not None] if not outs: return items # no exam possible → keep (best-effort) diff --git a/backend/board_artefacts.py b/backend/board_artefacts.py index 5305172..13e8d8b 100644 --- a/backend/board_artefacts.py +++ b/backend/board_artefacts.py @@ -1,10 +1,11 @@ """Board 2 „Artefakte": per finished block, prepare the learning artefacts the guide presents. A card is spawned by board 1's `done` column per mirrored block and runs through: - subblocks → facts → levels → relevance → question_pattern → artefacts → finalize + subblocks → facts → levels → relevance → question_pattern (+artefacts parallel) → finalize finalize (SERIAL) merges the block's results into the global sidecar/facts/pattern/artefakte -files + the DB tables. `outline` is a topic-wide BARRIER singleton at the very end -(prerequisite graph → chapter order), re-run once per generation run. +files + the DB tables. Danach zwei topic-weite BARRIEREN: `konsolidierung` (cross-block +sub dedup, faltet per repair.falte_sub) und `outline` (prerequisite graph → chapter order), +re-run once per generation run. The heavy lifting is the existing per-block functions in blocks.py — each card gets its own work subdirectory + facts/artefakte paths, so their slot files never collide across blocks.""" @@ -274,61 +275,56 @@ def _cross_schema(data) -> dict[int, str] | None: async def _proc_konsolidierung(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards): - """BARRIER/drain — cross-block sub dedup: the SAME statement carried by two blocks - (measured on Markdown: tab handling in 3 blocks, HTML blocks, backslash escapes — the - in-block paths never see these). Embedding candidates (block≠block, cos ≥ + """BARRIER/drain am RUN-ENDE — cross-block sub dedup: the SAME statement carried by two + blocks (measured on Markdown: tab handling in 3 blocks, HTML blocks, backslash escapes — + the in-block paths never see these). Embedding candidates (block≠block, cos ≥ SUB_DUP_KANDIDAT_COS) go to a two-judge panel; UNANIMITY decides which block keeps the - statement. The loser leaves its card's raw/facts and turns DB `variant` — before - questions/artefacts exist, so no orphans. Fail-open on judge failure/dissent.""" + statement. Sitzt seit dem Umbau NACH finalize: als Mittel-Barriere wartete jede fertige + Karte auf die langsamste (gemessen: 8:46 min Leerlauf pro Block, kanban-smoke). Der + Verlierer wird per repair.falte_sub gefaltet (variant + Fragen/Artefakte umhängen) — + die wenigen Cross-Dubletten kosten so ein paar umsonst generierte Artefakte statt + Minuten Wandzeit für alle. Fail-open on judge failure/dissent.""" + from repair import falte_sub topic = flow.topic work_dir = flow.work_dir - package_norms = {c["card_id"] for c in cards} - entries: list[tuple[int, str, str]] = [] # (card idx, block title, sub title); idx -1 = context - for ci, c in enumerate(cards): - for bt, subs in (c["payload"].get("raw") or {}).items(): - for s in subs: - entries.append((ci, bt, s)) - n_pkg = len(entries) - # Context: consensus subs of blocks already PAST this barrier (late spawns via the - # gap-check feedback would otherwise never be compared). Context never folds — - # its card payload lives downstream (board-1 rule: confirmed context always wins). - for r in await db.list_subblocks(topic): - if r["status"] == "consensus" and r["block_norm"] not in package_norms: - entries.append((-1, r["block"], r["sub_title"])) - ctx_facts: dict[str, dict] = {} # facts of downstream cards (DB rows carry none yet) - for bc in await db.kanban_cards(topic, board=BOARD, kind="ablock"): - if bc["card_id"] not in package_norms: - for bt, fm in (bc["payload"].get("facts") or {}).items(): - ctx_facts[_norm_title(bt)] = fm + # Resume-Karten aus der alten Stage-Position (Barriere lag vor den Fragen): erst fertig + # generieren — die Barriere feuert erneut, wenn alle wieder hier sind. Direkt dedupen + # ginge schief: finalize würde den gefalteten Sub aus dem Karten-Sidecar re-spiegeln. + nachzuegler = [(c["card_id"], "question_pattern") for c in cards + if "pattern" not in c["payload"]] + if nachzuegler: + await db.kanban_advance_many(topic, BOARD, nachzuegler) + flow.wake.set() + return async def _advance_all(): - await db.kanban_advance_many(topic, BOARD, [(c["card_id"], "question_pattern") for c in cards]) + await db.kanban_advance_many(topic, BOARD, [(c["card_id"], DONE) for c in cards]) flow.wake.set() - if n_pkg < 1 or len(entries) < 2 or not EMBEDDING_AKTIV or not await asyncio.to_thread(embedding.available): + rows = [r for r in await db.list_subblocks(topic) if r["status"] == "consensus"] + if len(rows) < 2 or not EMBEDDING_AKTIV or not await asyncio.to_thread(embedding.available): await _advance_all() return - sims = await asyncio.to_thread(embedding.embed_sims, [s for _, _, s in entries]) + sims = await asyncio.to_thread(embedding.embed_sims, [r["sub_title"] for r in rows]) if sims is None: await _advance_all() return - negs = [_neg_set(s) for _, _, s in entries] - pairs = [(i, j) for i in range(len(entries)) for j in range(i + 1, len(entries)) - if entries[i][0] != entries[j][0] and negs[i] == negs[j] + negs = [_neg_set(r["sub_title"]) for r in rows] + pairs = [(i, j) for i in range(len(rows)) for j in range(i + 1, len(rows)) + if rows[i]["block_norm"] != rows[j]["block_norm"] and negs[i] == negs[j] and float(sims[i][j]) >= SUB_DUP_KANDIDAT_COS] if not pairs: await _advance_all() return - def _kp(ci: int, bt: str, s: str) -> list: - if ci < 0: - f = ctx_facts.get(_norm_title(bt)) or {} - else: - f = (cards[ci]["payload"].get("facts") or {}).get(bt) or {} - return (f.get(_norm_title(s)) or {}).get("key_points") or [] + def _kp(r: dict) -> list: + try: + return (json.loads(r.get("facts") or "{}")).get("key_points") or [] + except ValueError: + return [] - def _side(tag: str, ci: int, bt: str, s: str) -> str: - return f"{tag}: [Block: {bt}] {s}" + "".join(f"\n - {p}" for p in _kp(ci, bt, s)) + def _side(tag: str, r: dict) -> str: + return f"{tag}: [Block: {r['block']}] {r['sub_title']}" + "".join(f"\n - {p}" for p in _kp(r)) # chunked judging: ONE call over all pairs scaled its timeout past 50 min, and a hung # call blocked the barrier for the full window (measured on aak: 196 pairs, 2×54 min) @@ -338,7 +334,7 @@ async def _proc_konsolidierung(ctx: GenContext, flow: Flow, files: dict, instruc """Two judges (+ substitute, + tie-breaker) over one pair chunk → {local_k: verdict}; empty dict = fail-open (pairs stay).""" lines = "\n\n".join( - f"{k}.\n{_side('A', *entries[i])}\n{_side('B', *entries[j])}" + f"{k}.\n{_side('A', rows[i])}\n{_side('B', rows[j])}" for k, (i, j) in enumerate(chunk, 1)) h = hashlib.md5(lines.encode()).hexdigest()[:8] paths = [work_dir / f"sub-crossblock-{h}-j{j}.json" for j in (1, 2)] @@ -374,7 +370,7 @@ async def _proc_konsolidierung(ctx: GenContext, flow: Flow, files: dict, instruc disputed = [k for k, v in final.items() if v == "uneinig"] if disputed: # tie-breaker: a third judge sees ONLY the disputed pairs, majority 2/3 d_lines = "\n\n".join( - f"{x}.\n{_side('A', *entries[chunk[k - 1][0]])}\n{_side('B', *entries[chunk[k - 1][1]])}" + f"{x}.\n{_side('A', rows[chunk[k - 1][0]])}\n{_side('B', rows[chunk[k - 1][1]])}" for x, k in enumerate(disputed, 1)) p3 = work_dir / f"sub-crossblock-{h}-j3.json" await _judge(3, p3, d_lines, len(disputed)) @@ -397,39 +393,23 @@ async def _proc_konsolidierung(ctx: GenContext, flow: Flow, files: dict, instruc for k, v in fin.items(): final_all[cnr * CROSS_CHUNK_PAARE + k] = v journal = {"paare": len(pairs), "chunks": len(chunks), "gefaltet": [], "verdicts": []} - gone: set[int] = set() - touched: set[int] = set() + gone: set[tuple] = set() for k, (i, j) in enumerate(pairs, 1): verdict = final_all.get(k, "nein") - journal["verdicts"].append({"a": f"{entries[i][1]} · {entries[i][2]}", - "b": f"{entries[j][1]} · {entries[j][2]}", + journal["verdicts"].append({"a": f"{rows[i]['block']} · {rows[i]['sub_title']}", + "b": f"{rows[j]['block']} · {rows[j]['sub_title']}", "verdict": verdict}) if verdict not in ("a", "b"): continue - lose = j if verdict == "a" else i - if entries[lose][0] < 0: # context never folds — the package side goes instead - lose = i if lose == j else j - keep = i if lose == j else j - if lose in gone or keep in gone: # keeper already folded → don't chain away the content + win, lose = (rows[i], rows[j]) if verdict == "a" else (rows[j], rows[i]) + wk = (win["block_norm"], win["sub_norm"]) + lk = (lose["block_norm"], lose["sub_norm"]) + if lk in gone or wk in gone: # keeper already folded → don't chain away the content continue - ci, bt, s = entries[lose] - p = cards[ci]["payload"] - if s in (p.get("raw") or {}).get(bt, []): - p["raw"][bt].remove(s) - (p.get("facts") or {}).get(bt, {}).pop(_norm_title(s), None) - sc = (p.get("sidecar") or {}).get(bt) - if isinstance(sc, list): # questions/artefacts consume the sidecar downstream - p["sidecar"][bt] = [e for e in sc - if _norm_title(str((e or {}).get("title", ""))) != _norm_title(s)] - await db.set_subblock_fields(topic, _norm_title(bt), _norm_title(s), status="variant") - gone.add(lose) - touched.add(ci) - journal["gefaltet"].append({"weg": f"{bt} · {s}", - "bleibt": f"{entries[keep][1]} · {entries[keep][2]}"}) - for ci in touched: - p = cards[ci]["payload"] - p["raw"] = {bt: subs for bt, subs in (p.get("raw") or {}).items() if subs} - await db.kanban_set_payload(topic, BOARD, cards[ci]["card_id"], p) + await falte_sub(topic, files, win, lose) + gone.add(lk) + journal["gefaltet"].append({"weg": f"{lose['block']} · {lose['sub_title']}", + "bleibt": f"{win['block']} · {win['sub_title']}"}) if journal["gefaltet"]: _log(topic, f"Sub-Crossblock: {len(journal['gefaltet'])} blockübergreifende Dublette(n) gefaltet") hg = hashlib.md5("\n".join(f"{i}:{j}" for i, j in pairs).encode()).hexdigest()[:8] @@ -483,25 +463,33 @@ async def _proc_relevance(ctx: GenContext, flow: Flow, files: dict, instructions sub["relevance"] = rel.get(gid, "relevant") p["sidecar"] = sidecar await db.kanban_set_payload(topic, BOARD, norm, p) - await db.kanban_advance(topic, BOARD, norm, "konsolidierung") + await db.kanban_advance(topic, BOARD, norm, "question_pattern") await _gather_cards(ctx, flow, cards, one) async def _proc_question_pattern(ctx: GenContext, flow: Flow, files: dict, instructions: str, cards): + """Fragen UND Artefakte im Fächer: beide brauchen nur den sidecar, nichts voneinander — + als Stage-Treppe kosteten sie zwei serielle Call-Segmente auf dem kritischen Pfad. + Die artefacts-Stage bleibt für Resume-Karten alter Läufe registriert.""" topic = flow.topic async def one(c): p = c["payload"] norm = c["card_id"] - pattern = await _question_pattern_block(ctx, _card_set_p(flow, norm), _pfiles(files, norm), - p.get("sidecar") or {}, instructions, - ns=f"{_safe(norm)}-", lbl=f"{p.get('title', norm)} · ") + pattern, artefacts = await asyncio.gather( + _question_pattern_block(ctx, _card_set_p(flow, norm), _pfiles(files, norm), + p.get("sidecar") or {}, instructions, + ns=f"{_safe(norm)}-", lbl=f"{p.get('title', norm)} · "), + _artefacts_block(ctx, _card_set_p(flow, norm), _pfiles(files, norm), + p.get("sidecar") or {}, instructions, + ns=f"{_safe(norm)}-", lbl=f"{p.get('title', norm)} · ")) if pattern is None: return _fail_or_cancel(ctx, f"Fragen {p.get('title', norm)}") p["pattern"] = pattern + p["artefacts"] = artefacts or {} # artefacts are optional — never fatal await db.kanban_set_payload(topic, BOARD, norm, p) - await db.kanban_advance(topic, BOARD, norm, "artefacts") + await db.kanban_advance(topic, BOARD, norm, "finalize") await _gather_cards(ctx, flow, cards, one) @@ -594,7 +582,7 @@ async def _proc_finalize(ctx: GenContext, flow: Flow, files: dict, cards): data = json.dumps({k: v for k, v in e.items() if k not in ("block", "subblock")}, ensure_ascii=False) await db.put_sub_artifact(topic, bnorm, sn, typ, data, bt, str(e.get("subblock", ""))) - await db.kanban_advance(topic, BOARD, c["card_id"], DONE) + await db.kanban_advance(topic, BOARD, c["card_id"], "konsolidierung") _log(topic, f"Artefakte fertig: {title}") flow.wake.set() @@ -647,15 +635,17 @@ def artefact_stages(ctx: GenContext, flow: Flow, files: dict, q: dict, folder, Stage(BOARD, "facts", lambda cs: _proc_facts(ctx, flow, files, q, folder, instructions, cs)), Stage(BOARD, "levels", lambda cs: _proc_levels(ctx, flow, files, instructions, cs)), Stage(BOARD, "relevance", lambda cs: _proc_relevance(ctx, flow, files, instructions, cs)), - # Barrier sits AFTER the sub-local stages: cards used to idle here median 36 min - # while levels/relevance work was still ahead of them + Stage(BOARD, "question_pattern", + lambda cs: _proc_question_pattern(ctx, flow, files, instructions, cs)), + # Resume-Pfad: Karten alter Läufe, die noch in artefacts stehen + Stage(BOARD, "artefacts", lambda cs: _proc_artefacts(ctx, flow, files, instructions, cs)), + Stage(BOARD, "finalize", lambda cs: _proc_finalize(ctx, flow, files, cs), serial=True), + # Cross-Block-Dedup als END-Barriere: als Mittel-Barriere idelte jede fertige Karte + # auf die langsamste (8:46 min/Block gemessen); jetzt faltet sie nach finalize + # per repair.falte_sub — spät gefundene Dubletten kosten Artefakt-Tokens, keine Wandzeit Stage(BOARD, "konsolidierung", lambda cs: _proc_konsolidierung(ctx, flow, files, instructions, cs), barrier=True, drain=True), - Stage(BOARD, "question_pattern", - lambda cs: _proc_question_pattern(ctx, flow, files, instructions, cs)), - Stage(BOARD, "artefacts", lambda cs: _proc_artefacts(ctx, flow, files, instructions, cs)), - Stage(BOARD, "finalize", lambda cs: _proc_finalize(ctx, flow, files, cs), serial=True), Stage(BOARD, "outline", lambda cs: _proc_outline(ctx, flow, files, instructions, cs), barrier=True, drain=True, gate=research_done), ] diff --git a/backend/board_inventory.py b/backend/board_inventory.py index 3d47b82..a740c1b 100644 --- a/backend/board_inventory.py +++ b/backend/board_inventory.py @@ -1699,19 +1699,23 @@ async def run_boards(ctx: GenContext, set_p, files: dict, q: dict, folder, instr 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) + try: + await kanban.run_flow(flow, stages, [_as_producer(p) for p in producers], set_p) + finally: + stopper.cancel() + if watcher: + watcher.cancel() + if ctx.is_cancelled(): + return False + if flow.state.get("qa_paused"): + return True # kein Fehler: Flow hielt am QA-Gate, Karten warten in subblocks + await _write_final(topic, files) + await _write_run_summary(topic, flow) + return True finally: - stopper.cancel() - if watcher: - watcher.cancel() + # erst NACH der Abschluss-QA leeren: deren Judge-Events gehören zum Lauf — + # vorher fielen sie ohne run_id aus jeder Run-Aggregation (Lauf 20260704-1452-b223) db.set_current_run(topic, None) - 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 @@ -1741,7 +1745,7 @@ async def _qa_gate_watch(ctx: GenContext, flow: Flow, inv_names: list[str], set_ 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) + await qa.write_report(report) except Exception: log.exception("[%s] QA-Report schreiben fehlgeschlagen", topic) if note >= QA_GATE_NOTE: @@ -1781,7 +1785,7 @@ async def _write_run_summary(topic: str, flow: Flow): summary["note"] = report["note"] summary["note_artefakte"] = report.get("note_artefakte") summary["artefakte"] = report.get("artefakte", {}) - await asyncio.to_thread(qa._write_report, report) + await qa.write_report(report) except Exception: log.exception("[%s] Abschluss-QA fehlgeschlagen", topic) atomic_write_json(flow.work_dir / "lauf-summary.json", summary, indent=1) diff --git a/backend/config.py b/backend/config.py index 3284acc..3f51166 100644 --- a/backend/config.py +++ b/backend/config.py @@ -194,6 +194,11 @@ KANBAN_BATCH = 5 # cards a worker pulls per micro-batch MAX_CARD_RETRIES = 3 # failures per card → dead-letter RETRY_BACKOFF = 30.0 # base seconds; backoff = base · 2^(retries-1) MAX_RESTARTS = 2 # agent restart cap per race slot +# Stall-Hedge: läuft ein Race-Slot so lange ohne Ergebnis, startet parallel ein Zwilling +# (key -h), der erste valide gewinnt. Gemessen (kanban-smoke): 4 Panel-Stalls à 160–230 s +# verlängerten den kritischen Pfad um ~5 min — gesunde Judge-Calls liegen bei p90 ≤ 105 s. +# 0 = aus. +HEDGE_NACH_S = 90 JUDGE_CHUNK = 40 # repair: findings per judge call EVIDENCE_PER_BLOCK = 6000 # repair: excerpt chars per fremd candidate ABSCHLUSS_QA_LLM = 1 # 0 = Abschluss-QA ohne LLM-Judges (Training misst selbst; spart Minuten) diff --git a/backend/guide_board.py b/backend/guide_board.py index e7dcbfc..5802874 100644 --- a/backend/guide_board.py +++ b/backend/guide_board.py @@ -618,73 +618,75 @@ async def run_guide_board(guide_id: str, topic: str, format_name: str, entries: import uuid from datetime import datetime, timezone db.set_current_run(topic, f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M')}-g{uuid.uuid4().hex[:4]}") - spec = (TEMPLATES_DIR / "Format" / "Section.md").read_text(encoding="utf-8") - subs_raw = await _load_subblocks(topic) - project = source_folder(topic) - fallback = (_prompt("Guide-Facts-Projekt", project=project) if project - else _prompt("Guide-Facts-Thema")) - env = _Env(ctx, guide_id, topic, format_name, instructions, content_path, - subs_raw, await _chapter_map(topic, entries), fallback, spec) - for num, line in entries.items(): - title = _title(line) - await db.upsert_guide_card(topic, format_name, _norm_title(title), title) - cards = await db.list_guide_cards(topic, format_name) - open_cards = [c for c in cards if c["stage"] != "done"] - if open_cards: - sem = asyncio.Semaphore(CARD_CONCURRENCY) + try: + spec = (TEMPLATES_DIR / "Format" / "Section.md").read_text(encoding="utf-8") + subs_raw = await _load_subblocks(topic) + project = source_folder(topic) + fallback = (_prompt("Guide-Facts-Projekt", project=project) if project + else _prompt("Guide-Facts-Thema")) + env = _Env(ctx, guide_id, topic, format_name, instructions, content_path, + subs_raw, await _chapter_map(topic, entries), fallback, spec) + for num, line in entries.items(): + title = _title(line) + await db.upsert_guide_card(topic, format_name, _norm_title(title), title) + cards = await db.list_guide_cards(topic, format_name) + open_cards = [c for c in cards if c["stage"] != "done"] + if open_cards: + sem = asyncio.Semaphore(CARD_CONCURRENCY) - async def _progress(): - while True: - counts = await db.guide_stage_counts(topic, format_name) - done = counts.get("done", 0) - total = sum(counts.values()) - await db.update_guide(guide_id, progress=f"Board: {done}/{total} Karten fertig") - await asyncio.sleep(2.0) + async def _progress(): + while True: + counts = await db.guide_stage_counts(topic, format_name) + done = counts.get("done", 0) + total = sum(counts.values()) + await db.update_guide(guide_id, progress=f"Board: {done}/{total} Karten fertig") + await asyncio.sleep(2.0) - reporter = asyncio.create_task(_progress()) - try: - await asyncio.gather(*[_run_card(env, c, sem) for c in open_cards]) - finally: - reporter.cancel() - db.set_current_run(topic, None) - else: + reporter = asyncio.create_task(_progress()) + try: + await asyncio.gather(*[_run_card(env, c, sem) for c in open_cards]) + finally: + reporter.cancel() + if is_guide_cancelled(guide_id): + return None + # assembly — identical shape to the legacy pipeline + cards = await db.list_guide_cards(topic, format_name) + chapters: list[dict] = [] + by_chapter: dict[str, list[dict]] = {} + order: list[str] = [] + for c in sorted(cards, key=lambda c: (c["ord"], c["block_norm"])): + if c["stage"] != "done": + _log(topic, f"Guide: Karte '{c['block']}' nicht fertig ({c['stage']}) — Abschnitt fehlt") + continue + sec = _first_section(c["md"]) + if sec is None: + continue + ch = c["chapter"] or "Inhalte" + if ch not in by_chapter: + by_chapter[ch] = [] + order.append(ch) + by_chapter[ch].append({ + "num": c["ord"], "title": c["block"], "md": sec["md"], + "compact": sec.get("compact", ""), "anchor": sec.get("anchor", ""), + "anker_compact": sec.get("anker_compact", ""), "subs": sec.get("subs", []), + "checkable": format_name == "Guide" or bool( + any(s.get("relevance") == "relevant" for s in subs_raw.get(c["block"], []))), + }) + for ch in order: + chapters.append({"title": ch, "sections": by_chapter[ch]}) + if chapters: + try: # Abschluss-Guide-QA (best-effort): speist das Badge mit einer frischen Note + import guide_qa + rep = await guide_qa.guide_qa_report(topic, llm=True) + if rep: + await asyncio.to_thread(guide_qa._write_report, rep) + except Exception: + log.exception("[%s] Abschluss-Guide-QA fehlgeschlagen", topic) + return chapters or None + finally: + # erst NACH der Abschluss-Guide-QA leeren: deren Judge-Events gehören zum + # Lauf — vorher fielen sie ohne run_id aus jeder Run-Aggregation db.set_current_run(topic, None) - if is_guide_cancelled(guide_id): - return None - # assembly — identical shape to the legacy pipeline - cards = await db.list_guide_cards(topic, format_name) - chapters: list[dict] = [] - by_chapter: dict[str, list[dict]] = {} - order: list[str] = [] - for c in sorted(cards, key=lambda c: (c["ord"], c["block_norm"])): - if c["stage"] != "done": - _log(topic, f"Guide: Karte '{c['block']}' nicht fertig ({c['stage']}) — Abschnitt fehlt") - continue - sec = _first_section(c["md"]) - if sec is None: - continue - ch = c["chapter"] or "Inhalte" - if ch not in by_chapter: - by_chapter[ch] = [] - order.append(ch) - by_chapter[ch].append({ - "num": c["ord"], "title": c["block"], "md": sec["md"], - "compact": sec.get("compact", ""), "anchor": sec.get("anchor", ""), - "anker_compact": sec.get("anker_compact", ""), "subs": sec.get("subs", []), - "checkable": format_name == "Guide" or bool( - any(s.get("relevance") == "relevant" for s in subs_raw.get(c["block"], []))), - }) - for ch in order: - chapters.append({"title": ch, "sections": by_chapter[ch]}) - if chapters: - try: # Abschluss-Guide-QA (best-effort): speist das Badge mit einer frischen Note - import guide_qa - rep = await guide_qa.guide_qa_report(topic, llm=True) - if rep: - await asyncio.to_thread(guide_qa._write_report, rep) - except Exception: - log.exception("[%s] Abschluss-Guide-QA fehlgeschlagen", topic) - return chapters or None async def done_step(topic: str, format_name: str) -> int: diff --git a/backend/pipeline.py b/backend/pipeline.py index 819cb1f..7ef04b1 100644 --- a/backend/pipeline.py +++ b/backend/pipeline.py @@ -164,10 +164,18 @@ _relevance_schema = _enum_map_schema("relevance", _RELEVANCE) # relevance ∈ _yesno_schema = _enum_map_schema("relevant", _YESNO) # triage gate ∈ ja/nein -from config import MAX_RESTARTS as _MAX_RESTARTS # noqa: E402 — zentral tunebar +from config import MAX_RESTARTS as _MAX_RESTARTS, HEDGE_NACH_S as _HEDGE_NACH_S # noqa: E402 — zentral tunebar + +# Detached Nachzügler-Tasks (late-Fold): Referenz gegen GC, Aufräumen via done-callback. +_NACHZUEGLER: set[asyncio.Task] = set() -async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: int, provider: str, on_update=None, cancelled=None, *, grace: int | None = None, min_runtime: int | None = None, max_runtime: int | None = None) -> list | None: +def _detached(task: asyncio.Task) -> None: + _NACHZUEGLER.add(task) + task.add_done_callback(_NACHZUEGLER.discard) + + +async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: int, provider: str, on_update=None, cancelled=None, *, grace: int | None = None, min_runtime: int | None = None, max_runtime: int | None = None, late=None) -> list | None: """Starts all slots in parallel and collects `quorum` valid results. Slot spec: {key, prompt, role, capabilities, payload}. `payload(result)` @@ -185,24 +193,59 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: elapses while agents are still running — gives them time to search thoroughly. `max_runtime` (wall-clock from start): hard cap — returns whatever is collected (or None if nothing), killing the rest. Both default off; only Research sets them. + + `late(value)` (async): Nachzügler werden beim Quorum-Return NICHT gekillt, sondern + laufen detached weiter; jedes noch eintreffende valide Ergebnis geht an `late`. + Ersetzt den grace-Timer der Finder-Runden — der hielt die Runde bis 300 s offen, + nur damit die dritte Stimme zählt (gemessen: 73 s Warten pro Runde). """ attempts = {i: 0 for i in range(len(slots))} tasks: dict[asyncio.Task, int] = {} + keys: dict[asyncio.Task, str] = {} + born: dict[asyncio.Task, float] = {} + hedged: set[int] = set() # slot got its one twin — no hedge cascades + fertig: set[int] = set() # slot delivered a valid result (late twins are ignored) loop = asyncio.get_running_loop() start = loop.time() min_deadline = start + min_runtime if min_runtime else None max_deadline = start + max_runtime if max_runtime else None deadline: float | None = None - def spawn(i: int) -> None: + def spawn(i: int, suffix: str = "") -> None: slot = slots[i] lbl = slot.get("label") or (label if len(slots) == 1 else f"{label} {i + 1}") + key = slot["key"] + suffix task = asyncio.create_task(run_agent( - slot["key"], slot["prompt"], timeout, + key, slot["prompt"], timeout, provider=provider, role=slot["role"], capabilities=slot["capabilities"], scope=topic, on_line=slot.get("on_line"), label=lbl, )) tasks[task] = i + keys[task] = key + born[task] = loop.time() + + spaet: set[int] = set() # je Slot zählt nur EIN spätes Ergebnis (Hedge-Zwilling = Echo) + + def _detach_rest() -> None: + """Quorum steht: Nachzügler an `late` übergeben statt killen (nur Erfolgs-Return).""" + if late is None: + return + for t, i in list(tasks.items()): + tasks.pop(t) + keys.pop(t, None) + born.pop(t, None) + + async def _warte(t=t, i=i): + try: + r = await t + if i in spaet: + return + if r and r[0] == 0 and (val := slots[i]["payload"](r)) is not None: + spaet.add(i) + await late(val) + except (asyncio.CancelledError, Exception): # noqa: BLE001 — Nachzügler sind best-effort + pass + _detached(asyncio.create_task(_warte())) for i in range(len(slots)): spawn(i) @@ -218,8 +261,20 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: return results or None min_ok = min_deadline is None or loop.time() >= min_deadline if deadline is not None and len(results) >= quorum and loop.time() >= deadline and min_ok: + _detach_rest() return results - # Wake up for the earliest relevant deadline (grace, min, or max). + # Hedge: a slot running HEDGE_NACH_S without result gets ONE parallel twin + # (key -h) — first valid result wins. Stalled provider calls burned the full + # timeout cap before the restart even began (measured: 160–230 s per stall). + if _HEDGE_NACH_S: + now = loop.time() + for t in [t for t in list(tasks) if tasks[t] not in hedged | fertig + and now - born[t] >= _HEDGE_NACH_S]: + i = tasks[t] + hedged.add(i) + spawn(i, suffix="-h") + _log(topic, f"{label} {i + 1}: {_HEDGE_NACH_S}s ohne Ergebnis — Hedge-Zwilling gestartet") + # Wake up for the earliest relevant deadline (grace, min, max, or next hedge). waits = [] if deadline is not None and len(results) >= quorum: waits.append(deadline - loop.time()) @@ -227,12 +282,21 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: waits.append(min_deadline - loop.time()) if max_deadline is not None: waits.append(max_deadline - loop.time()) + if _HEDGE_NACH_S: + naechste = [born[t] + _HEDGE_NACH_S - loop.time() for t in tasks + if tasks[t] not in hedged | fertig] + if naechste: + waits.append(max(0.0, min(naechste))) wait_timeout = max(0.0, min(waits)) if waits else None done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED, timeout=wait_timeout) if not done: continue for task in done: i = tasks.pop(task) + keys.pop(task, None) + born.pop(task, None) + if i in fertig: + continue # späte Zwillinge eines bereits gewerteten Slots payload, err = None, None try: result = task.result() @@ -249,6 +313,10 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: if payload is not None: results.append(payload) + fertig.add(i) + for t2 in [t2 for t2, i2 in tasks.items() if i2 == i]: # Zwilling killen + kill_process(keys.get(t2, slots[i]["key"])) + t2.cancel() if grace is not None and deadline is None: deadline = loop.time() + grace _log(topic, f"{label}: first result — grace {grace}s running") @@ -256,23 +324,26 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: on_update(len(results)) if (len(results) >= quorum and (grace is None or loop.time() >= deadline) and (min_deadline is None or loop.time() >= min_deadline)): + _detach_rest() return results continue _log(topic, f"{label} {i + 1} (attempt {attempts[i] + 1}): {err}") attempts[i] += 1 # If the minimum already stands, restarts are pointless — the restart - # would be killed at the grace end anyway. + # would be killed at the grace end anyway. A still-running twin IS the retry. enough = grace is not None and len(results) >= quorum - if attempts[i] <= _MAX_RESTARTS and not enough and not (cancelled and cancelled()): + zwilling = any(i2 == i for i2 in tasks.values()) + if attempts[i] <= _MAX_RESTARTS and not enough and not zwilling and not (cancelled and cancelled()): spawn(i) if len(results) >= quorum: # all slots done, minimum stands (only reachable with grace) + _detach_rest() return results _log(topic, f"{label}: quorum {quorum} not reached ({len(results)} valid)") return None finally: for task, i in tasks.items(): - kill_process(slots[i]["key"]) + kill_process(keys.get(task, slots[i]["key"])) task.cancel() if tasks: await asyncio.gather(*tasks.keys(), return_exceptions=True) diff --git a/backend/qa.py b/backend/qa.py index e0dfb2f..ed8a862 100644 --- a/backend/qa.py +++ b/backend/qa.py @@ -412,6 +412,21 @@ def _write_report(report: dict) -> Path: return path +async def write_report(report: dict) -> Path: + """_write_report + kompaktes kind='qa'-Event. Die Report-JSONs liegen nur auf der + Lauf-Maschine (storage/qa/) — ein DB-Pull reichte nicht, um Note/Quoten eines Runs + zu rekonstruieren (Analyse 20260704-1452-b223). Nur die Kennzahlen, kein Volltext; + run_id stempelt add_event aus der Registry (gesetzt im Lauf, leer bei manueller QA).""" + path = await asyncio.to_thread(_write_report, report) + try: # Event ist Komfort — ein DB-Fehler darf den Report nicht kosten (fail-open) + await db.add_event(report["topic"], "qa", key=path.stem, meta={ + "note": report["note"], "note_artefakte": report.get("note_artefakte"), + "quoten": report["quoten"], "quoten_artefakte": report.get("quoten_artefakte", {})}) + except Exception: + pass + return path + + def _digest(report: dict, path: Path): na = report.get("note_artefakte") print(f"QA {report['topic']} — {report['bloecke']} Blöcke (run {report['run_id'] or '—'})" @@ -443,7 +458,7 @@ async def main(topic: str, llm: bool): report = await qa_report(topic, llm=llm) if report is None: sys.exit(1) - _digest(report, _write_report(report)) + _digest(report, await write_report(report)) finally: await db.close_db() diff --git a/backend/repair.py b/backend/repair.py index 6a704ab..b2fc95c 100644 --- a/backend/repair.py +++ b/backend/repair.py @@ -6,7 +6,6 @@ deterministisch, bestätigte Dubletten mergen (Zweitmeinung), Fremd/Unecht nur n Gegen-Judge entfernen (fail-open: Zweifel/Fehler → behalten). Lücken brauchen Recherche, Verwaiste den nächsten Board-2-Lauf — beides wird nur ausgewiesen.""" -import asyncio import json import logging import re @@ -47,7 +46,7 @@ async def repair_befunde(topic: str) -> dict: # blendete sub_dubletten aus und ließ die Note zwischen 10.0 und ~9 pendeln neu = await qa.qa_report(topic, llm=True) if neu: - await asyncio.to_thread(qa._write_report, neu) + await qa.write_report(neu) return {"hygiene": hygiene, "merges": merges, "sub_merges": sub_merges, "entfernt": entfernt, "aufgeraeumt": aufgeraeumt, "braucht_research": len(report.get("luecken", []))} @@ -135,6 +134,31 @@ def _sub_gewinner(a: dict, b: dict) -> tuple[dict, dict]: return (a, b) if score(a) >= score(b) else (b, a) +async def falte_sub(topic: str, files: dict, win: dict, lose: dict) -> None: + """Verlierer-Sub falten: Status variant, Fragen/Artefakte zum Gewinner umhängen (oder + löschen, wenn der Typ dort existiert), Sidecar-Dateien bereinigen. Gemeinsamer Kern + von QA-Repair und Cross-Block-Dedup (Board 2, Run-Ende) — win/lose sind subblocks-Rows.""" + await db.set_subblock_fields(topic, lose["block_norm"], lose["sub_norm"], status="variant") + w_fragen = {r["sub_norm"] for r in await db.list_question_pattern(topic, win["block_norm"])} + for r in await db.list_question_pattern(topic, lose["block_norm"]): + if r["sub_norm"] != lose["sub_norm"]: + continue + if win["sub_norm"] not in w_fragen: + await db.upsert_question_pattern(topic, win["block_norm"], win["sub_norm"], + win["block"], win["sub_title"], r["question"]) + await db.delete_frage_row(topic, lose["block_norm"], lose["sub_norm"]) + w_typen = {r["type"] for r in await db.get_sub_artefakte(topic, block_norm=win["block_norm"]) + if r["sub_norm"] == win["sub_norm"]} + for r in await db.get_sub_artefakte(topic, block_norm=lose["block_norm"]): + if r["sub_norm"] != lose["sub_norm"]: + continue + if r["type"] not in w_typen: + await db.put_sub_artifact(topic, win["block_norm"], win["sub_norm"], r["type"], + r["data"], win["block"], win["sub_title"]) + await db.delete_artefakt_row(topic, lose["block_norm"], lose["sub_norm"], r["type"]) + _entferne_sub_in_files(files, lose["block_norm"], lose["sub_norm"]) + + async def _merge_sub_dubletten(topic: str, report: dict, files: dict) -> list[str]: """QA-bestätigte Sub-Paare (llm=ja) nach Zweitmeinung falten: Verlierer → variant, seine Fragen/Artefakte wandern zum Gewinner (oder fallen weg, wenn er den Typ hat). @@ -161,27 +185,8 @@ async def _merge_sub_dubletten(topic: str, report: dict, files: dict) -> list[st wk, lk = (win["block_norm"], win["sub_norm"]), (lose["block_norm"], lose["sub_norm"]) if v.get(i) != "ja" or wk in gone or lk in gone: continue - await db.set_subblock_fields(topic, lose["block_norm"], lose["sub_norm"], status="variant") + await falte_sub(topic, files, win, lose) gone.add(lk) - # Fragen/Artefakte des Verlierers: umhängen, wenn der Gewinner den Typ nicht hat - w_fragen = {r["sub_norm"] for r in await db.list_question_pattern(topic, win["block_norm"])} - for r in await db.list_question_pattern(topic, lose["block_norm"]): - if r["sub_norm"] != lose["sub_norm"]: - continue - if win["sub_norm"] not in w_fragen: - await db.upsert_question_pattern(topic, win["block_norm"], win["sub_norm"], - win["block"], win["sub_title"], r["question"]) - await db.delete_frage_row(topic, lose["block_norm"], lose["sub_norm"]) - w_typen = {r["type"] for r in await db.get_sub_artefakte(topic, block_norm=win["block_norm"]) - if r["sub_norm"] == win["sub_norm"]} - for r in await db.get_sub_artefakte(topic, block_norm=lose["block_norm"]): - if r["sub_norm"] != lose["sub_norm"]: - continue - if r["type"] not in w_typen: - await db.put_sub_artifact(topic, win["block_norm"], win["sub_norm"], r["type"], - r["data"], win["block"], win["sub_title"]) - await db.delete_artefakt_row(topic, lose["block_norm"], lose["sub_norm"], r["type"]) - _entferne_sub_in_files(files, lose["block_norm"], lose["sub_norm"]) merged.append(f"{lose['sub_title'][:40]} → {win['sub_title'][:40]}") return merged diff --git a/backend/routes.py b/backend/routes.py index 28247cb..6855519 100644 --- a/backend/routes.py +++ b/backend/routes.py @@ -175,7 +175,7 @@ async def run_qa_route(req: QaRunRequest): report = await qa.qa_report(req.topic, llm=req.llm) if report is None: raise HTTPException(status_code=404, detail="keine fertigen Bausteine") - await asyncio.to_thread(qa._write_report, report) + await qa.write_report(report) note_guide = None try: # fertiger Guide vorhanden → mitmessen (ein Klick, drei Noten); best-effort import guide_qa diff --git a/backend/tests/test_board_inventory.py b/backend/tests/test_board_inventory.py index deb006b..e85c192 100644 --- a/backend/tests/test_board_inventory.py +++ b/backend/tests/test_board_inventory.py @@ -176,6 +176,38 @@ async def test_board1_full_flow(board_env): assert summary["boards"].get("inventory", {}).get("done_block") == 4 +async def test_abschluss_qa_events_tragen_run_id(board_env, monkeypatch): + """Abschluss-QA läuft NACH run_flow — ihre Judge-Events müssen trotzdem die run_id + des Laufs tragen (Lauf 20260704-1452-b223: run_id leer → aus jeder Aggregation gefallen).""" + import asyncio + + import qa as qa_mod + db, ctx, files = board_env + await _seed(db) + + async def qa_mit_judge_event(topic, llm=False): + # wie die echten LLM-Judges: run_agent schreibt ein agent-Event + await db.add_event(topic, "agent", key=f"qa-{topic}-bausteine-0", status="ok") + return {"note": 10.0, "topic": topic, "quoten": {}, "fremd": [], + "artefakte": {"status": "nicht generiert"}} + monkeypatch.setattr(qa_mod, "qa_report", qa_mit_judge_event) + ok = await asyncio.wait_for( + bi.run_boards(ctx, lambda *a, **k: None, files, {"type": "thema"}, None, "", research=False), + timeout=30) + assert ok + summary = json.loads((files["arbeit"] / "lauf-summary.json").read_text(encoding="utf-8")) + conn = await db.get_db() + rows = await (await conn.execute( + "SELECT run_id FROM events WHERE topic=? AND key=?", + (TOPIC, f"qa-{TOPIC}-bausteine-0"))).fetchall() + assert rows and all(r[0] == summary["run_id"] for r in rows) + # Registry nach dem Lauf geleert: manuelle QA bleibt korrekt ohne run_id + await db.add_event(TOPIC, "agent", key="qa-manuell", status="ok") + row = await (await conn.execute( + "SELECT run_id FROM events WHERE topic=? AND key='qa-manuell'", (TOPIC,))).fetchone() + assert row[0] == "" + + async def test_filter_judges_run_parallel(board_env, monkeypatch): """40 Blöcke → 2 Filter-Chunks: die Judge-Welle muss parallel laufen (Perf-Fix).""" import asyncio diff --git a/backend/tests/test_konsolidierung.py b/backend/tests/test_konsolidierung.py index 8082fa8..080ca9f 100644 --- a/backend/tests/test_konsolidierung.py +++ b/backend/tests/test_konsolidierung.py @@ -388,85 +388,88 @@ class _FakeEmb: return arr @ arr.T -async def _cross_env(db, tmp_path): +async def _cross_env(db, tmp_path, finalisiert=True): + """Zwei finalisierte Karten in der End-Barriere; die Sub-Rows liegen in der DB + (post-finalize ist die DB die Wahrheit, nicht mehr das Karten-Payload).""" flow = Flow(TOPIC, work_dir=tmp_path) cards = [] for bnorm, subs in (("alpha", ["Gleiche Aussage", "Nur in Alpha"]), ("beta", ["Gleiche Aussage", "Nur in Beta"])): - payload = {"title": bnorm.title(), - "raw": {bnorm.title(): list(subs)}, - "sidecar": {bnorm.title(): [{"title": s, "level": "beginner"} for s in subs]}, - "facts": {bnorm.title(): {blocks._norm_title(s): {"key_points": [f"kp {s}"]} for s in subs}}} + payload = {"title": bnorm.title()} + if finalisiert: + payload.update(pattern={}, artefacts={}) await db.kanban_upsert_card(TOPIC, "artefacts", bnorm, "ablock", "konsolidierung", payload) await _seed_block(db, bnorm, subs) cards.append({"card_id": bnorm, "payload": payload}) - return flow, cards + files = {k: tmp_path / f"{k}.json" for k in + ("sub_roh", "facts", "sidecar", "question_pattern", "artefakte")} + return flow, cards, files async def test_crossblock_folds_loser(testdb, tmp_path, monkeypatch): - """Einstimmig „a" → Beta verliert die geteilte Aussage, Karten wandern zu levels.""" + """Einstimmig „a" → Betas geteilte Aussage wird variant, ihre Frage wandert zum + Gewinner (falte_sub), Karten gehen auf DONE.""" db = testdb - flow, cards = await _cross_env(db, tmp_path) + flow, cards, files = await _cross_env(db, tmp_path) + sn = blocks._norm_title("Gleiche Aussage") + await db.upsert_question_pattern(TOPIC, "beta", sn, "Beta", "Gleiche Aussage", "F?") monkeypatch.setattr(ba, "embedding", _FakeEmb) fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "j2": {"pairs": {"1": "a"}}}) monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) assert "Gleiche Aussage" in fake.calls[0]["prompt"] - beta = await db.kanban_get_card(TOPIC, "artefacts", "beta") - assert beta["stage"] == "question_pattern" - assert beta["payload"]["raw"]["Beta"] == ["Nur in Beta"] - assert blocks._norm_title("Gleiche Aussage") not in beta["payload"]["facts"]["Beta"] - # Barriere liegt jetzt hinter levels/relevance → auch die sidecar muss den Fold tragen - assert [e["title"] for e in beta["payload"]["sidecar"]["Beta"]] == ["Nur in Beta"] - alpha = await db.kanban_get_card(TOPIC, "artefacts", "alpha") - assert alpha["stage"] == "question_pattern" - assert alpha["payload"]["raw"]["Alpha"] == ["Gleiche Aussage", "Nur in Alpha"] - rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} - assert rows[blocks._norm_title("Gleiche Aussage")] == "variant" + for cid in ("alpha", "beta"): + assert (await db.kanban_get_card(TOPIC, "artefacts", cid))["stage"] == ba.DONE + beta_rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert beta_rows[sn] == "variant" + alpha_rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "alpha")} + assert alpha_rows[sn] == "consensus" + fragen = await db.list_question_pattern(TOPIC) + assert {(r["block_norm"], r["sub_norm"]) for r in fragen} == {("alpha", sn)} # umgehängt async def test_crossblock_tiebreaker_folds(testdb, tmp_path, monkeypatch): """j1/j2 uneinig → j3 entscheidet mit Mehrheit; hier „a" → Beta verliert.""" db = testdb - flow, cards = await _cross_env(db, tmp_path) + flow, cards, files = await _cross_env(db, tmp_path) monkeypatch.setattr(ba, "embedding", _FakeEmb) fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "j2": {"pairs": {"1": "nein"}}, "j3": {"pairs": {"1": "a"}}}) monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) assert len(fake.calls) == 3 - beta = await db.kanban_get_card(TOPIC, "artefacts", "beta") - assert beta["payload"]["raw"]["Beta"] == ["Nur in Beta"] + rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert rows[blocks._norm_title("Gleiche Aussage")] == "variant" async def test_crossblock_dissent_without_tiebreaker_keeps_both(testdb, tmp_path, monkeypatch): """j3 liefert nichts (FAILED) → fail-open, Paar bleibt.""" db = testdb - flow, cards = await _cross_env(db, tmp_path) + flow, cards, files = await _cross_env(db, tmp_path) monkeypatch.setattr(ba, "embedding", _FakeEmb) fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "j2": {"pairs": {"1": "b"}}}) # j3 fehlt → FAILED monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) - beta = await db.kanban_get_card(TOPIC, "artefacts", "beta") - assert beta["stage"] == "question_pattern" - assert beta["payload"]["raw"]["Beta"] == ["Gleiche Aussage", "Nur in Beta"] + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) + assert (await db.kanban_get_card(TOPIC, "artefacts", "beta"))["stage"] == ba.DONE + rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert rows[blocks._norm_title("Gleiche Aussage")] == "consensus" async def test_crossblock_ersatzrichter(testdb, tmp_path, monkeypatch): """Nur ein Richter liefert → Ersatz jE als zweite Stimme; Einstimmigkeit faltet.""" db = testdb - flow, cards = await _cross_env(db, tmp_path) + flow, cards, files = await _cross_env(db, tmp_path) monkeypatch.setattr(ba, "embedding", _FakeEmb) fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "jE": {"pairs": {"1": "a"}}}) # j2 → FAILED monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) - beta = await db.kanban_get_card(TOPIC, "artefacts", "beta") - assert beta["payload"]["raw"]["Beta"] == ["Nur in Beta"] + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) + rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert rows[blocks._norm_title("Gleiche Aussage")] == "variant" async def test_crossblock_without_embedding_advances(testdb, tmp_path, monkeypatch): db = testdb - flow, cards = await _cross_env(db, tmp_path) + flow, cards, files = await _cross_env(db, tmp_path) class _Aus: @staticmethod @@ -478,33 +481,27 @@ async def test_crossblock_without_embedding_advances(testdb, tmp_path, monkeypat monkeypatch.setattr(ba, "embedding", _Aus) monkeypatch.setattr(ba, "run_single_slot", kein_agent) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) + for cid in ("alpha", "beta"): + assert (await db.kanban_get_card(TOPIC, "artefacts", cid))["stage"] == ba.DONE + + +async def test_crossblock_nachzuegler_zurueck_zu_fragen(testdb, tmp_path, monkeypatch): + """Resume-Karte aus der alten Stage-Position (kein pattern im Payload) → zurück nach + question_pattern, KEIN Dedup — finalize würde den Fold sonst re-spiegeln.""" + db = testdb + flow, cards, files = await _cross_env(db, tmp_path, finalisiert=False) + + async def kein_agent(*a, **kw): + raise AssertionError("Nachzügler dürfen keinen Dedup auslösen") + + monkeypatch.setattr(ba, "embedding", _FakeEmb) + monkeypatch.setattr(ba, "run_single_slot", kein_agent) + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) for cid in ("alpha", "beta"): assert (await db.kanban_get_card(TOPIC, "artefacts", cid))["stage"] == "question_pattern" - - -async def test_crossblock_context_wins(testdb, tmp_path, monkeypatch): - """Kontext-Sub (Block schon hinter der Barrier) gewinnt auch bei Verdict „b" — - die Paket-Seite fällt, der Kontext bleibt unangetastet.""" - db = testdb - flow = Flow(TOPIC, work_dir=tmp_path) - payload = {"title": "Alpha", "raw": {"Alpha": ["Gleiche Aussage"]}, "facts": {"Alpha": {}}} - await db.kanban_upsert_card(TOPIC, "artefacts", "alpha", "ablock", "konsolidierung", payload) - await _seed_block(db, "alpha", ["Gleiche Aussage"]) - cards = [{"card_id": "alpha", "payload": payload}] - # Kontext-Block "gamma" ist bereits weiter (Stage levels) und hält dieselbe Aussage - await db.kanban_upsert_card(TOPIC, "artefacts", "gamma", "ablock", "levels", - {"title": "Gamma", "raw": {"Gamma": ["Gleiche Aussage"]}, "facts": {}}) - await _seed_block(db, "gamma", ["Gleiche Aussage"]) - monkeypatch.setattr(ba, "embedding", _FakeEmb) - # Verdict „a": das Paket (A) soll behalten — Kontext faltet trotzdem nie - fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "j2": {"pairs": {"1": "a"}}}) - monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) - alpha = await db.kanban_get_card(TOPIC, "artefacts", "alpha") - assert alpha["payload"]["raw"].get("Alpha", []) == [] # Paket-Seite gefaltet - gamma_rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "gamma")} - assert gamma_rows[blocks._norm_title("Gleiche Aussage")] == "consensus" # Kontext unberührt + rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert rows[blocks._norm_title("Gleiche Aussage")] == "consensus" async def test_finalize_defaultet_level_nachzuegler(testdb, tmp_path): @@ -547,21 +544,20 @@ async def test_crossblock_chunking_faltet_global(testdb, tmp_path, monkeypatch): cards = [] for bnorm, subs in (("alpha", ["Gleiche Aussage", "Zweite gleiche Aussage", "Nur in Alpha"]), ("beta", ["Gleiche Aussage", "Zweite gleiche Aussage"])): - payload = {"title": bnorm.title(), - "raw": {bnorm.title(): list(subs)}, - "sidecar": {bnorm.title(): [{"title": s, "level": "beginner"} for s in subs]}, - "facts": {bnorm.title(): {blocks._norm_title(s): {"key_points": []} for s in subs}}} + payload = {"title": bnorm.title(), "pattern": {}, "artefacts": {}} await db.kanban_upsert_card(TOPIC, "artefacts", bnorm, "ablock", "konsolidierung", payload) await _seed_block(db, bnorm, subs) cards.append({"card_id": bnorm, "payload": payload}) + files = {k: tmp_path / f"{k}.json" for k in + ("sub_roh", "facts", "sidecar", "question_pattern", "artefakte")} monkeypatch.setattr(ba, "embedding", _FakeEmb) monkeypatch.setattr(ba, "CROSS_CHUNK_PAARE", 1) # 2 Paare → 2 Chunks fake = _fake_slot({"j1": {"pairs": {"1": "a"}}, "j2": {"pairs": {"1": "a"}}}) monkeypatch.setattr(ba, "run_single_slot", fake) - await ba._proc_konsolidierung(_ctx(), flow, {}, "", cards) + await ba._proc_konsolidierung(_ctx(), flow, files, "", cards) assert len(fake.calls) == 4 # 2 Chunks × j1/j2 - beta = await db.kanban_get_card(TOPIC, "artefacts", "beta") - assert beta["payload"]["raw"].get("Beta", []) == [] # beide Dubletten global gefaltet + rows = {r["sub_norm"]: r["status"] for r in await db.list_subblocks(TOPIC, "beta")} + assert set(rows.values()) == {"variant"} # beide Dubletten global gefaltet async def test_facts_nachfass_holt_nur_fehlende(testdb, tmp_path, monkeypatch): diff --git a/backend/tests/test_qa.py b/backend/tests/test_qa.py index 2aecf03..aa35d5f 100644 --- a/backend/tests/test_qa.py +++ b/backend/tests/test_qa.py @@ -1,5 +1,7 @@ """QA-Detektoren: Fehler-Injektion auf Mini-Korpus — deterministisch, ohne LLM/Embedding.""" +import json + import qa @@ -238,6 +240,25 @@ async def test_unecht_braucht_doppelt_nein(testdb, tmp_path, monkeypatch): assert report["unecht"] == ["Wackelkandidat"] +async def test_write_report_spiegelt_note_als_event(testdb, tmp_path, monkeypatch): + """Report-JSONs liegen nur auf der Lauf-Maschine — write_report spiegelt Note/Quoten + als kind='qa'-Event in die DB, damit ein DB-Pull für die Run-Analyse reicht.""" + db = testdb + monkeypatch.setattr(qa, "QA_DIR", tmp_path) + report = {"topic": "t", "run_id": "20260704-1452-b223", "note": 9.3, "note_artefakte": 8.0, + "quoten": {"luecken": 0.1}, "quoten_artefakte": {"verwaiste": 0.0}} + path = await qa.write_report(report) + assert path.stem == "20260704-1452-b223" + conn = await db.get_db() + row = await (await conn.execute( + "SELECT key, meta, run_id FROM events WHERE topic='t' AND kind='qa'")).fetchone() + assert row and row[0] == "20260704-1452-b223" + meta = json.loads(row[1]) + assert meta["note"] == 9.3 and meta["note_artefakte"] == 8.0 + assert meta["quoten"] == {"luecken": 0.1} and meta["quoten_artefakte"] == {"verwaiste": 0.0} + assert row[2] == "" # manuelle QA ohne Lauf → leeres run_id ist korrekt + + async def test_topic_delete_entfernt_qa_ordner(testdb, tmp_path, monkeypatch): """DELETE /topics räumt auch storage/qa// — Reports gehören zum Topic.""" import routes diff --git a/backend/tests/test_race.py b/backend/tests/test_race.py new file mode 100644 index 0000000..6fc4c96 --- /dev/null +++ b/backend/tests/test_race.py @@ -0,0 +1,138 @@ +"""_race-Hedging: Stall-Slots bekommen einen parallelen Zwilling statt den Timeout-Cap +abzuwarten (gemessen: 4 Panel-Stalls à 160–230 s pro Lauf auf dem kritischen Pfad).""" + +import asyncio + +import pipeline + + +def _slot(payload=lambda r: r[1]): + return {"key": "k1", "prompt": "p", "role": "judge", "capabilities": "none", "payload": payload} + + +async def test_hedge_zwilling_rettet_stall(monkeypatch): + """Original stallt → nach HEDGE_NACH_S startet der Zwilling (key -h), sein Ergebnis + gewinnt, das hängende Original wird gekillt.""" + calls, killed = [], [] + + async def fake_agent(key, prompt, timeout, **kw): + calls.append(key) + if key.endswith("-h"): + return (0, "zwilling", "") + await asyncio.sleep(30) # Stall — würde sonst den ganzen Cap verbrennen + return (0, "original", "") + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "kill_process", lambda k: killed.append(k)) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0.05) + res = await pipeline._race("t", "Test", [_slot()], 1, 60, "claude") + assert res == ["zwilling"] + assert calls == ["k1", "k1-h"] + assert "k1" in killed # das hängende Original läuft nicht weiter + + +async def test_hedge_original_gewinnt_zwilling_wird_gekillt(monkeypatch): + """Kommt das Original doch noch vor dem Zwilling an, wird der Zwilling gekillt + und sein spätes Ergebnis nicht gewertet.""" + killed = [] + + async def fake_agent(key, prompt, timeout, **kw): + await asyncio.sleep(0.3 if key.endswith("-h") else 0.15) + return (0, key, "") + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "kill_process", lambda k: killed.append(k)) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0.05) + res = await pipeline._race("t", "Test", [_slot()], 1, 60, "claude") + assert res == ["k1"] + assert "k1-h" in killed + + +async def test_hedge_aus_bei_null(monkeypatch): + """HEDGE_NACH_S=0 → kein Zwilling, Verhalten wie zuvor.""" + calls = [] + + async def fake_agent(key, prompt, timeout, **kw): + calls.append(key) + await asyncio.sleep(0.1) + return (0, "ok", "") + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0) + res = await pipeline._race("t", "Test", [_slot()], 1, 60, "claude") + assert res == ["ok"] + assert calls == ["k1"] + + +async def test_late_fold_nachzuegler_zaehlt_nach(monkeypatch): + """Quorum 2 kehrt sofort zurück; der dritte Slot wird nicht gekillt, sein Ergebnis + geht an `late` (ersetzt den grace-Timer der Finder-Runden).""" + import time + killed, spaet = [], [] + + async def fake_agent(key, prompt, timeout, **kw): + if key == "k3": + await asyncio.sleep(0.2) + return (0, "dritter", "") + return (0, key, "") + + async def late(val): + spaet.append(val) + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "kill_process", lambda k: killed.append(k)) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0) + slots = [{"key": f"k{i}", "prompt": "p", "role": "quick", "capabilities": "none", + "payload": lambda r: r[1]} for i in (1, 2, 3)] + t0 = time.monotonic() + res = await pipeline._race("t", "Test", slots, 2, 60, "claude", late=late) + assert time.monotonic() - t0 < 0.15 # kein Warten auf k3 + assert sorted(res) == ["k1", "k2"] + assert "k3" not in killed + await asyncio.sleep(0.3) + assert spaet == ["dritter"] + + +async def test_late_fold_invalider_nachzuegler_ignoriert(monkeypatch): + """Nachzügler mit invalidem Payload löst late NICHT aus (best-effort).""" + spaet = [] + + async def fake_agent(key, prompt, timeout, **kw): + if key == "k3": + await asyncio.sleep(0.1) + return (1, "", "kaputt") + return (0, key, "") + + async def late(val): + spaet.append(val) + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "kill_process", lambda k: None) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0) + slots = [{"key": f"k{i}", "prompt": "p", "role": "quick", "capabilities": "none", + "payload": lambda r: r[1]} for i in (1, 2, 3)] + res = await pipeline._race("t", "Test", slots, 2, 60, "claude", late=late) + assert res is not None + await asyncio.sleep(0.25) + assert spaet == [] + + +async def test_hedge_zwilling_ersetzt_restart(monkeypatch): + """Scheitert das Original, während der Zwilling noch läuft, gibt es KEINEN + zusätzlichen Restart — der Zwilling ist der Retry.""" + calls = [] + + async def fake_agent(key, prompt, timeout, **kw): + calls.append(key) + if key.endswith("-h"): + await asyncio.sleep(0.2) + return (0, "zwilling", "") + await asyncio.sleep(0.1) + return (1, "", "kaputt") # Fehler NACH dem Hedge-Start + + monkeypatch.setattr(pipeline, "run_agent", fake_agent) + monkeypatch.setattr(pipeline, "kill_process", lambda k: None) + monkeypatch.setattr(pipeline, "_HEDGE_NACH_S", 0.05) + res = await pipeline._race("t", "Test", [_slot()], 1, 60, "claude") + assert res == ["zwilling"] + assert calls == ["k1", "k1-h"] # kein dritter Spawn diff --git a/backend/tests/test_subblocks.py b/backend/tests/test_subblocks.py index c70daeb..84af8e5 100644 --- a/backend/tests/test_subblocks.py +++ b/backend/tests/test_subblocks.py @@ -57,7 +57,7 @@ def _mk_race(finder_by_agent): prompts = [] async def fake_race(topic, label, slots, quorum, timeout, provider, on_update=None, - cancelled=None, *, grace=None, min_runtime=None, max_runtime=None): + cancelled=None, *, grace=None, min_runtime=None, max_runtime=None, late=None): outs = [] for slot in slots: key, prompt = slot["key"], slot["prompt"] @@ -372,6 +372,47 @@ async def test_clarify_inline_evidence_no_tools(sub_env, monkeypatch, tmp_path): assert list(tmp_path.glob("subblock-final-*-j*.md")) # Engine persistiert die Judge-Antwort +async def test_panel_2of3_kehrt_bei_einigkeit_zurueck(tmp_path): + """Zwei übereinstimmende Verdicts → Rückkehr ohne den langsamen Dritten; sein + Ergebnis wird detached nachpersistiert (Resume).""" + import asyncio as aio + gesunken = {} + + async def judge(j, delay, antwort): + await aio.sleep(delay) + return (0, antwort, "") + + tasks = {aio.create_task(judge(1, 0.01, "a")): 1, + aio.create_task(judge(2, 0.02, "a")): 2, + aio.create_task(judge(3, 5.0, "b")): 3} + + def sink(j, r): + gesunken[j] = r[1] + + import time + t0 = time.monotonic() + await blx._panel_2of3(tasks, sink, lambda: list(gesunken.values()), lambda s: s) + assert time.monotonic() - t0 < 1.0 # nicht auf j3 gewartet + assert gesunken == {1: "a", 2: "a"} + + +async def test_panel_2of3_dissens_wartet_auf_dritten(): + """Uneinige erste zwei → der dritte wird abgewartet (Mehrheit braucht ihn).""" + import asyncio as aio + gesunken = {} + + async def judge(j, delay, antwort): + await aio.sleep(delay) + return (0, antwort, "") + + tasks = {aio.create_task(judge(1, 0.01, "a")): 1, + aio.create_task(judge(2, 0.02, "b")): 2, + aio.create_task(judge(3, 0.1, "a")): 3} + await blx._panel_2of3(tasks, lambda j, r: gesunken.__setitem__(j, r[1]), + lambda: list(gesunken.values()), lambda s: s) + assert gesunken == {1: "a", 2: "b", 3: "a"} + + async def test_facts_check_inline_cited_evidence(sub_env, monkeypatch, tmp_path): """Facts-Check: zitierte Zeilenbereiche gehen inline mit, Judge läuft ohne Tools, die Check-Datei schreibt die Engine aus der Text-Antwort.""" @@ -409,6 +450,35 @@ async def test_facts_check_inline_cited_evidence(sub_env, monkeypatch, tmp_path) assert (tmp_path / f"facts-check-{sh}-c0-j1.json").exists() # Engine persistiert die Antwort +async def test_facts_find_inline_evidence_no_tools(sub_env, monkeypatch, tmp_path): + """Facts find/erg mit Korpus: Auszüge inline, Agent ohne Tools — Tool-Agenten + verloren sich in Reasoning-Schleifen und endeten mit leerem Turn (Retry-Wellen).""" + db, ctx, files = sub_env + d = _corpus(tmp_path) + seen = [] + facts = {"facts": [{"block": "Alpha", "subblock": "Sub Eins", "key_points": ["k"], + "prerequisites": "", "hurdles": "", + "cited_facts": [{"text": "T", "source": "Skript.txt, Z.2"}], + "example_idea": ""}]} + + async def fake_slot(ctx2, label, *, key, prompt, role, capabilities, payload, timeout, on_line=None): + seen.append((key, capabilities, prompt)) + return blx.OK, payload((0, json.dumps(facts), "")) + + async def fake_agent(key, prompt, timeout, **kw): # Check-Panel + return (0, '{"ok": true}', "") + + monkeypatch.setattr(blx, "run_single_slot", fake_slot) + monkeypatch.setattr(blx, "run_agent", fake_agent) + res = await blx._facts_block(ctx, lambda *a, **k: None, {"arbeit": tmp_path}, + {"Alpha": ["Sub Eins"]}, {"type": "uni"}, d, "", ns="x-") + assert res is not None + finder = [s for s in seen if "-facts-c0" in s[0] or "-facts-erg-" in s[0]] + assert finder and all(caps == "none" for _, caps, _ in finder) + assert all("── Skript.txt" in prompt for _, _, prompt in finder) # Auszüge inline + assert all("ls/find" not in prompt for _, _, prompt in finder) + + def test_sub_key_resolves_short_titles(): """Artefakt-Agenten echoen den Kurztitel; der Sub-Key heißt 'kurztitel: beschreibung'. Eindeutiger Präfix wird aufgelöst, Mehrdeutiges und Fehlendes bleibt unverändert."""