"""Blocks pipeline: research consensus + clarification loop — pure inventory, unsorted. 5x research (min. 3, grace) → mapping (consensus/rest) → clarification loop (max. CONSENSUS_MAX_ROUNDS rounds): 3 selection agents (min. 2, grace) decide on the disputed rest, a mapping agent sorts into accept/discard/ still disputed. An empty rest ends the loop; the last round must decide everything. Races use a grace window instead of "first N win": after the first valid result, the remaining agents get CONSENSUS_GRACE seconds to finish. The consensus is accumulated in code — no agent re-emits the full list. """ import asyncio import json import logging import math import re import shutil import subprocess import time from pathlib import Path import database as db import embedding from agents import kill_process, cancel_scope, clear_scope, run_agent from config import CONSENSUS_GRACE, RESEARCH_GRACE, CONSENSUS_MAX_ROUNDS, DEFAULT_PROVIDER, CRAWL_KEEP_PATTERNS, CRAWL_NOISE_PATTERNS, CRAWL_MIN_CHARS, QUELLE_RELEVANZ_CHUNK, QUELLE_RELEVANZ_SNIPPET, EMBEDDING_AKTIV, EMBEDDING_SUB_DUP from fsutil import atomic_write_text, atomic_write_json from jsonio import 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, _relevance_schema, _runde_schema, _semaphore, _str_list, _levels_schema, _timeout, run_single_slot, ) from textkit import ( _unique_title, _load_blocks, _norm_title, _parse_selection, _parse_subblocks, _title, _resolve_title, _title_index, ) # Chunk the subblocks (web search per block): 1 agent per ~10 blocks, capped. SUBBLOCK_CHUNK = 10 SUBBLOCK_MAX = 40 # Classifying is cheap (short verdict, no web search) → larger packages, fewer files/agents. LEVEL_CHUNK = 100 # Research: fixed file batches instead of a search loop → each crawl page is assigned exactly once. RESEARCH_BATCH = 20 # crawl pages per batch RESEARCH_READERS = 2 # reader agents per batch (consensus ≥2 within the batch) RESEARCH_THEMA_AGENTS = 5 # web mode (source "thema", no crawl folder) # uni/projekt: chunk the script text into sections of ~this size (against lost-in-the-middle on # large documents). ~12k chars ≈ 3k tokens → safely below the recall-drop threshold. RESEARCH_SECTION_CHARS = 12000 # Triage (content/noise) is now a deterministic rule filter (config.CRAWL_*). SUBBLOCK_CAP = 900 # subblock find loop per chunk (15 min) CONSOLIDATION_CHUNK = 600 # up to here ONE global judge (dedups everything); above that chunked + merge pass — fallback path only DEDUP_PAIR_FLOOR = 0.6 # min cosine for a candidate pair (complete-link aggregates → no chaining) DEDUP_PAIRS_CHUNK = 40 # pairs per judge package (pairwise verification instead of a block mixer) FILTER_CHUNK = 35 # blocks to assess per judge in the degrade pass (full list as context) # Balance question-pattern chunks by sub load via LPT (makespan), not by block count. QUESTION_CHUNK_SUBS = 50 # target sum of relevant subs per chunk QUESTION_MAX_ROUNDS = 3 # catch-up rounds for subs without a pattern (the LLM omits ~18 % per chunk) FACTS_CHUNK_SUBS = 25 # facts extraction: smaller chunks (facts are bulkier than patterns) FACTS_CHECK_PANEL = 3 # judges per chunk in the facts check (majority objects) CONSOLIDATION_PANEL = 3 # mapping judges per chunk (panel → reconcile instead of a single judge) SUBBLOCK_PANEL = 3 # source judges in the subblock clarification (majority instead of a single judge) log = logging.getLogger("creator.blocks") _blocks_progress: dict[str, str] = {} _blocks_errors: dict[str, str] = {} _blocks_cancelled: set[str] = set() _blocks_step: dict[str, int] = {} ARTEFACT_TYPES = ("flashcard", "example") def load_source(topic: str) -> dict: """Read the persisted source choice. Fallback (legacy topics without source.json): if projects/ exists → projekt, otherwise thema.""" q = _json_file(source_path(topic)) if isinstance(q, dict) and q.get("type") in ("thema", "projekt", "uni", "link"): return q if project_dir(topic).is_dir(): return {"type": "projekt", "location": f"projects/{topic}", "spec": ""} return {"type": "thema", "location": "", "spec": ""} def source_folder(topic: str) -> Path | None: """Folder source (projekt/uni → path, link → crawl folder) — otherwise None (thema).""" q = load_source(topic) if q["type"] == "link": return source_crawl_dir(topic) if q["type"] in ("projekt", "uni"): return safe_folder(q.get("location", "")) return None def _crawl_done(topic: str) -> bool: return (source_crawl_dir(topic) / ".done").exists() # marker only on clean completion # Learning-path levels (beginner/advanced/expert); old difficulty values are backward-compatible. _LEVELS = ("beginner", "advanced", "expert", "easy", "medium", "hard") async def subblocks_title(topic: str, block: str) -> list[str]: """Subblock titles of a block — DB-first (consensus), fallback to the sidecar file.""" rows = [s["sub_title"] for s in await db.list_subblocks(topic, _norm_title(block)) if s["status"] == "consensus" and s["sub_title"]] if rows: return rows sc = _json_file(subblocks_path(topic)) if not isinstance(sc, dict): return [] return [ t for s in (sc.get(block) or []) if isinstance(s, dict) and (t := str(s.get("title", "")).strip()) ] async def load_question_pattern(topic: str, block: str) -> list[dict]: """Predefined question patterns of a block — DB-first, fallback to sidecar (empty = live).""" rows = await db.list_question_pattern(topic, _norm_title(block)) if rows: return [{"subblock": r["sub_title"], "question": r["question"]} for r in rows if r["question"]] fm = _json_file(question_pattern_path(topic)) if not isinstance(fm, dict): return [] return [ {"subblock": str(e.get("subblock", "")).strip(), "question": question} for e in (fm.get(block) or []) if isinstance(e, dict) and (question := str(e.get("question", "")).strip()) ] async def subblocks_frei(topic: str, block: str, max_level: int) -> list[str]: """Subblock titles up to the unlocked level (≤ max_level). Fallback without level knowledge (legacy/sidecar): all subblock titles.""" rows = await db.subs_with_level(topic, block) if not rows: return await subblocks_title(topic, block) return [s["title"] for s in rows if s["level"] <= max_level and s["title"]] async def load_question_pattern_free(topic: str, block: str, max_level: int) -> list[dict]: """Question patterns, filtered to subblocks up to the unlocked level. Without level knowledge (legacy/sidecar), unfiltered.""" rows = await db.subs_with_level(topic, block) if not rows: return await load_question_pattern(topic, block) unlocked = {s["norm"] for s in rows if s["level"] <= max_level} return [m for m in await load_question_pattern(topic, block) if _norm_title(m["subblock"]) in unlocked] async def load_overview(topic: str) -> list[dict]: """Structured block list for the overview — DB-first (consensus + subs/levels/relevance), fallback to blocks.md + sidecar (legacy topics).""" bs = await db.list_blocks(topic, status="consensus") if bs: out = [] for num, b in enumerate(bs, 1): subs = [s for s in await db.list_subblocks(topic, b["title_norm"]) if s["status"] == "consensus"] out.append({ "num": num, "title": b["title"], "description": b["description"], "subblocks": [ {"title": s["sub_title"], "level": s["level"] if s["level"] in _LEVELS else "advanced", "relevance": s["relevance"] if s["relevance"] in ("relevant", "peripheral") else None} for s in subs if s["sub_title"] ], }) return out entries = _load_blocks(_read(blocks_path(topic))) sidecar = _json_file(subblocks_path(topic)) sidecar = sidecar if isinstance(sidecar, dict) else {} out = [] for num, entry in entries.items(): title = _title(entry) split_parts = entry.split(" — ", 1) description = split_parts[1].strip() if len(split_parts) == 2 else "" subblocks = [ { "title": t, "level": s.get("level") if s.get("level") in _LEVELS else "advanced", "relevance": s.get("relevance") if s.get("relevance") in ("relevant", "peripheral") else None, } for s in (sidecar.get(title) or []) if isinstance(s, dict) and (t := str(s.get("title", "")).strip()) ] out.append({"num": num, "title": title, "description": description, "subblocks": subblocks}) return out def _blocks_steps(topic: str) -> tuple: """Steps per source: link gets "Source laden" up front, projekt additionally "Supplement". Subblocks + levels are three phases each (find, select, clarify). Per phase all packages run in parallel; the step remains until the last package is done. """ q = load_source(topic) base = ("Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter") rest = ( "Subblocks find", "Subblocks select", "Subblocks clarify", "Facts find", "Facts check", "Facts fix", "Levels find", "Levels select", "Levels clarify", "Relevance find", "Relevance select", "Relevance clarify", "Outline", "Questions find", "Questions select", "Questions clarify", "Questions check", "Flashcards", "Examples", ) middle = base + (("Supplement",) if q["type"] == "projekt" else ()) + rest return (("Source prep",) if q["type"] == "link" else ()) + middle def _step_idx(topic: str, name: str) -> int: return _blocks_steps(topic).index(name) def _report_p(set_p, topic: str, step: str): """Async report callback for _gather_progress: sets " d/t…" + step index.""" idx = _step_idx(topic, step) async def report(d, t): set_p(f"{step} {d}/{t}…", step=idx) return report # Coarse display phases: bundle the fine steps (internally everything stays fine-grained). # Special steps (Source laden, Supplement) belong to the "Inventory" phase. PHASEN = ( ("Source", ("Source prep",)), ("Inventory", ("Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter", "Supplement")), ("Subblocks", ("Subblocks find", "Subblocks select", "Subblocks clarify")), ("Facts", ("Facts find", "Facts check", "Facts fix")), ("Levels", ("Levels find", "Levels select", "Levels clarify")), ("Relevance", ("Relevance find", "Relevance select", "Relevance clarify")), ("Outline", ("Outline",)), ("Questions", ("Questions find", "Questions select", "Questions clarify", "Questions check")), ("Artefacts", ("Flashcards", "Examples")), ) def _phases(topic: str) -> list[tuple[str, int]]: """[(coarse_label, number of present fine steps)] for the current source.""" fine_steps = _blocks_steps(topic) return [(label, n) for label, members in PHASEN if (n := sum(f in members for f in fine_steps))] def _phases_status(topic: str, current: int | None) -> list[dict]: """Coarse phase states from the fine progress `current` (None = all pending, len(feine) = all done). → [{label, state}] with state done/active/pending.""" out, start = [], 0 for label, n in _phases(topic): end = start + n if current is None or current < start: state = "pending" elif current >= end: state = "done" else: state = "active" out.append({"label": label, "state": state}) start = end return out def _blocks_files(topic: str) -> dict: work_dir = arbeit_dir(topic) rounds = range(1, CONSENSUS_MAX_ROUNDS + 1) return { "final": blocks_path(topic), "arbeit": work_dir, "research": [work_dir / f"research-{i}.md" for i in (1, 2, 3, 4, 5)], "research_mapping": work_dir / "research-mapping.json", "selection": {n: [work_dir / f"selection-r{n}-{i}.json" for i in (1, 2, 3)] for n in rounds}, "mapping": {n: work_dir / f"selection-mapping-r{n}.json" for n in rounds}, "ergaenzung": work_dir / "ergaenzung.json", "sub_roh": work_dir / "subblocks-roh.json", "facts": work_dir / "subblocks-facts.json", "sidecar": subblocks_path(topic), "question_pattern": question_pattern_path(topic), "outline": work_dir / "outline.json", "outline_slots": [work_dir / f"outline-{i}.json" for i in (1, 2, 3)], "artefakte": work_dir / "artefakte.json", } def _all_slot_files(files: dict) -> list[Path]: work_dir = files["arbeit"] # Subblock/levels slots are dynamic per chunk — collect via glob. dyn = (list(work_dir.glob("subblock-*")) + list(work_dir.glob("facts-*")) + list(work_dir.glob("level-*")) + list(work_dir.glob("relevance-*")) + list(work_dir.glob("question-pattern-*")) + list(work_dir.glob("outline-*")) + list(work_dir.glob("artifact-*")) + list(work_dir.glob("research-*")) + list(work_dir.glob("consolidation-*")) + list(work_dir.glob("clarification*")) + list(work_dir.glob("dedup-*")) + list(work_dir.glob("inventar-filter*"))) if work_dir.is_dir() else [] return [ *files["research"], files["research_mapping"], *(p for slots in files["selection"].values() for p in slots), *files["mapping"].values(), files["ergaenzung"], files["sub_roh"], files["sidecar"], files["question_pattern"], files["facts"], files["outline"], files["artefakte"], *dyn, ] def cancel_blocks(topic: str) -> bool: if topic not in _blocks_progress: return False _blocks_cancelled.add(topic) cancel_scope(f"blocks-{topic}-") # waiting agents bail before spawning kill_process(f"blocks-{topic}-") # kill running subprocesses return True async def _resume_step(topic: str) -> int: """First step still open. While blocks.md is missing (never built OR a reset deleted it) the inventory sub-step comes fine-grained from the DB step status; once blocks.md exists the inventory counts as done (the artefact is the source of truth, robust for legacy topics) and later phases come from the persisted artefacts. A reset-from-inventory deletes blocks.md, so this stays exact.""" files = _blocks_files(topic) steps_all = _blocks_steps(topic) if not files["final"].exists(): for step in ("Source prep", "Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"): if step in steps_all and await db.get_step_status(topic, step) != "done": return _step_idx(topic, step) return _step_idx(topic, "Blocks-Filter") # statuses done but artefact gone → rewrite q = load_source(topic) if q["type"] == "projekt" and not files["ergaenzung"].exists(): return _step_idx(topic, "Supplement") sidecar = _json_file(files["sidecar"]) if _sidecar_schema(sidecar) is not None: # Levels done; only relevance still open? if not _relevance_complete(sidecar): return _step_idx(topic, "Relevance find") # Relevance done; outline (blocks artifact for the guide) open? if not _outline_complete(files): return _step_idx(topic, "Outline") # Outline done; question patterns open? if not _question_pattern_complete(topic): return _step_idx(topic, "Questions find") # Questions done; learning artefacts (flashcards/examples) open? if not _artefacts_complete(files): return _step_idx(topic, "Flashcards") return len(_blocks_steps(topic)) if _sub_raw_schema(_json_file(files["sub_roh"])) is None: return _step_idx(topic, "Subblocks find") # Subblocks done; facts still open? (Facts come before the levels.) if not _facts_complete(files): return _step_idx(topic, "Facts find") return _step_idx(topic, "Levels find") def _fine_status(topic: str, current: int | None) -> list[dict]: """Fine sub-step status: per step {label, phase, state}. state from `current` (done = idx). Phase label from PHASEN.""" step_phase = {s: label for label, steps in PHASEN for s in steps} out = [] for i, s in enumerate(_blocks_steps(topic)): state = "pending" if current is None or current < i else "done" if current > i else "active" out.append({"label": s, "phase": step_phase.get(s, ""), "state": state}) return out async def blocks_status(topic: str) -> dict: # Internally fine-grained (resume/progress); bundled into 5 coarse phases for display. fine_steps = _blocks_steps(topic) ready = blocks_path(topic).exists() # inventory written → block overview available generating = topic in _blocks_progress if generating: current = _blocks_step.get(topic) else: # True progress (inventory from DB step status, later phases from artefacts). current = await _resume_step(topic) partial = not generating and 0 < current < len(fine_steps) return { "ready": ready, "generating": generating, "progress": _blocks_progress.get(topic), "error": _blocks_errors.get(topic), "partial": partial, "steps": _phases_status(topic, current), "feine_steps": _fine_status(topic, current), } def active_blocks() -> list[dict]: return [{"topic": t, "progress": p} for t, p in _blocks_progress.items()] def reset_blocks(topic: str) -> None: """"Remove": deletes the ENTIRE blocks area — crawl, triage, inventory … questions. KEEPS only the topic config `source.json` (type/link/spec). Re-generating crawls anew. (Crawl/triage belong to the blocks; only the config is the "topic".)""" files = _blocks_files(topic) files["final"].unlink(missing_ok=True) files["sidecar"].unlink(missing_ok=True) files["question_pattern"].unlink(missing_ok=True) shutil.rmtree(source_crawl_dir(topic), ignore_errors=True) # crawl belongs to the blocks shutil.rmtree(files["arbeit"], ignore_errors=True) _blocks_errors.pop(topic, None) # source.json intentionally stays — that is the topic config. def _phase_idx(label: str) -> int: """Index of the coarse phase in the canonical order (Source=0 … Questions=5).""" order = [l for l, _ in PHASEN] return order.index(label) if label in order else 1 def _reset_from_phase(topic: str, label: str) -> None: """Delete file artefacts FROM the coarse phase `label` (Source/Inventory … Questions), keeping earlier ones. Cumulative. source.json + crawl (.done) always stay (re-crawl only on full reset).""" files = _blocks_files(topic) work_dir = files["arbeit"] idx = _phase_idx(label) def glob_del(pat: str) -> None: if work_dir.is_dir(): for p in work_dir.glob(pat): p.unlink(missing_ok=True) # Phase index: Source=0 · Inventory=1 · Subblocks=2 · Facts=3 · Levels=4 · Relevance=5 · Outline=6 · Questions=7 · Artefacts=8 if idx <= 8: # Artefacts (flashcards/examples) files["artefakte"].unlink(missing_ok=True) glob_del("artifact-*") if idx <= 7: # Questions files["question_pattern"].unlink(missing_ok=True) glob_del("question-pattern-*") if idx <= 6: # Outline files["outline"].unlink(missing_ok=True) glob_del("outline-*") if idx <= 5: # Relevance glob_del("relevance-*") if idx <= 4: # Levels + relevance share the sidecar → from Levels rebuild entirely files["sidecar"].unlink(missing_ok=True) glob_del("level-*") else: # from Relevance: keep levels, strip only the relevance fields sc = _json_file(files["sidecar"]) if isinstance(sc, dict): for subs in sc.values(): for s in (subs if isinstance(subs, list) else []): if isinstance(s, dict): s.pop("relevance", None) atomic_write_json(files["sidecar"], sc, indent=1) if idx <= 3: # Facts (before the levels) — facts map + work files gone files["facts"].unlink(missing_ok=True) glob_del("facts-*") if idx <= 2: # Subblocks files["sub_roh"].unlink(missing_ok=True) glob_del("subblock-*") if idx <= 1: # Inventory (and source) = inventory files + blocks.md gone for p_old in _all_slot_files(files): p_old.unlink(missing_ok=True) files["final"].unlink(missing_ok=True) async def _reset_from_step(topic: str, step_idx: int) -> None: """Fine reset FROM a sub-step (0-based index in _blocks_steps). Resets pipeline_state + artefacts + DB from here on; earlier steps stay. Inventory sub-steps reconstruct the DB status from the artefacts (dedup/filter safe; clarification robustly falls back to consolidation, because the clarification renames → a title mismatch would be fragile).""" fine_steps = list(_blocks_steps(topic)) if not (0 <= step_idx < len(fine_steps)): return affected = set(fine_steps[step_idx:]) files = _blocks_files(topic) work_dir = files["arbeit"] def gd(pat: str) -> None: if work_dir.is_dir(): for p in work_dir.glob(pat): p.unlink(missing_ok=True) await db.delete_pipeline_state(topic, list(fine_steps[step_idx:])) # Later artefacts/DB cumulatively from the affected step (back to front). if {"Examples", "Flashcards"} & affected: files["artefakte"].unlink(missing_ok=True); gd("artifact-*"); await db.delete_sub_artefakte(topic) if any(s.startswith("Questions") for s in affected): files["question_pattern"].unlink(missing_ok=True); gd("question-pattern-*"); await db.delete_question_pattern(topic) if "Outline" in affected: files["outline"].unlink(missing_ok=True); gd("outline-*"); await db.delete_outline(topic) if any(s.startswith("Relevance") for s in affected): gd("relevance-*") if any(s.startswith("Levels") for s in affected): gd("level-*") if any(s.startswith("Facts") for s in affected): files["facts"].unlink(missing_ok=True); gd("facts-*") # The sidecar carries subblocks + their fields level/relevance/facts. From subblocks rebuild entirely; # otherwise strip only the fields of the phases to rebuild — subblocks are preserved. if any(s.startswith("Subblock") for s in affected): files["sidecar"].unlink(missing_ok=True) else: strip = {f for s, f in (("Facts", "facts"), ("Levels", "level"), ("Relevance", "relevance")) if any(x.startswith(s) for x in affected)} if strip: sc = _json_file(files["sidecar"]) if isinstance(sc, dict): for subs in sc.values(): for s in (subs if isinstance(subs, list) else []): if isinstance(s, dict): for f in strip: s.pop(f, None) atomic_write_json(files["sidecar"], sc, indent=1) if any(s.startswith("Subblock") for s in affected): files["sub_roh"].unlink(missing_ok=True); gd("subblock-*"); await db.delete_subblocks(topic) # --- Inventory (DB status cascades) --- if "Blocks-Filter" in affected and not ({"Clarification", "Consolidation", "Research", "Dedup"} & affected): # Only filter rebuilt: degraded blocks back to consensus. d = _json_file(work_dir / "inventar-filter.json") for f in (d.get("fragments", []) if isinstance(d, dict) else []): await db.set_block_status(topic, _norm_title(f.get("fragment", "")), "consensus") gd("inventar-filter*") if "Dedup" in affected and not ({"Clarification", "Consolidation", "Research"} & affected): # Dedup (+filter) rebuilt: all blocks discarded in dedup/filter back to consensus. for kind in ("dedup-runde-1.json", "inventar-filter.json"): d = _json_file(work_dir / kind) title = ([t for g in d.get("groups", []) for t in g] if isinstance(d, dict) and "groups" in d else [f.get("fragment", "") for f in d.get("fragments", [])] if isinstance(d, dict) else []) for t in title: await db.set_block_status(topic, _norm_title(t), "consensus") gd("dedup-*"); gd("inventar-filter*") if {"Clarification", "Consolidation"} & affected and not ({"Research"} & affected): # Clarification/consolidation rebuilt: clear inventory DB (research readers stay). Clarification rollback # would be fragile due to renaming → cleanly rebuild from consolidation. await db.delete_blocks(topic) gd("clarification*"); gd("consolidation-*"); gd("dedup-*"); gd("inventar-filter*") if "Research" in affected: # whole inventory like a phase reset for p_old in _all_slot_files(files): p_old.unlink(missing_ok=True) await db.delete_blocks(topic) # blocks.md is the inventory aggregate — stale once any inventory sub-step is reset. Delete it so the # status/resume see the inventory as open from the reset step (the pipeline rewrites it; the DB step # statuses of the kept earlier steps let those skip). if {"Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"} & affected: files["final"].unlink(missing_ok=True) async def reset_blocks_ab_step(topic: str, step_idx: int) -> None: """Public: ONLY reset from a sub-step — no re-generation. Leaves a partial state (the steps from here count as open). If a generation is running → ignore.""" if topic in _blocks_progress: return await _reset_from_step(topic, step_idx) def _supplement_schema(data): """{"blocks": [{"title", "description"}]} → list (empty allowed) · otherwise None.""" if not isinstance(data, dict) or not isinstance(data.get("blocks"), list): return None out = [] for b in data["blocks"]: if not isinstance(b, dict) or not isinstance(b.get("title"), str) or not isinstance(b.get("description"), str): return None title, description = b["title"].strip(), b["description"].strip() if not title: return None out.append((title, description)) return out def _convert_pdfs(project: Path) -> None: """Convert PDFs in the project to .txt (pdftotext) — agents read text instead of page images. Called before every project generation; converts only if the .txt is missing or older than the PDF. The original is left untouched. If pdftotext is missing and the project contains PDFs → hard error instead of an unreliable direct-read mode (MiniMax image limit, vision cost). """ pdfs = list(project.rglob("*.pdf")) if not pdfs: return if shutil.which("pdftotext") is None: raise RuntimeError("pdftotext missing (install poppler-utils) — PDFs in the project cannot be read") for pdf in pdfs: txt = pdf.with_suffix(".txt") if txt.exists() and txt.stat().st_mtime >= pdf.stat().st_mtime: continue try: subprocess.run(["pdftotext", "-layout", str(pdf), str(txt)], check=True, timeout=120) _log(project.name, f"PDF converted: {pdf.name} → {txt.name}") except Exception as e: raise RuntimeError(f"PDF conversion failed ({pdf.name}): {e}") from e _SOURCE_TEMPLATE = {"projekt": "Blocks-Source-Projekt", "uni": "Blocks-Source-Uni", "link": "Blocks-Source-Link"} def _text_sections(text: str, goal: int = RESEARCH_SECTION_CHARS) -> list[str]: """Split text at paragraph/line boundaries into sections of ~`ziel` chars (against lost-in-the-middle on large documents). Small text stays ONE section. Content stays complete — only separating whitespace is dropped.""" text = text.strip() if len(text) <= goal: return [text] if text else [] sections: list[str] = [] buf = "" def flush(): nonlocal buf if buf.strip(): sections.append(buf.strip()) buf = "" for block in re.split(r"\n\s*\n", text): # at paragraph boundaries block = block.strip() if not block: continue if len(block) > goal: # single huge paragraph → hard-cut at lines flush() for line in block.split("\n"): if buf and len(buf) + len(line) + 1 > goal: flush() buf += line + "\n" flush() elif buf and len(buf) + len(block) + 2 > goal: flush() buf = block else: buf = (buf + "\n\n" + block) if buf else block flush() return sections def _build_research_prompt(topic: str, out_path: Path, instructions: str, type: str, folder: Path | None, fokus: str = "", section: str = "") -> str: if section: # Section mode (uni/projekt): text directly in the prompt → small context, no file reading. source = section elif type in _SOURCE_TEMPLATE: source = _prompt(_SOURCE_TEMPLATE[type], project=folder) else: source = _prompt("Blocks-Source-Thema", topic=topic) return _prompt( "Blocks-Research", topic=topic, source=source, blocks_path=out_path, focus=fokus, extra=_extra(instructions), ) def _file_payload(path: Path): """Valid if the slot file exists and contains numbered entries.""" if not path.exists(): return None text = path.read_text(encoding="utf-8") return text if _parse_selection(text) else None def _mapping_schema(data): """{"blocks": [str, ≥1], "rest": [str]} → (blocks, rest) · otherwise None.""" if not isinstance(data, dict): return None blocks = _str_list(data.get("blocks")) rest = _str_list(data.get("rest")) if not blocks or rest is None: return None return blocks, rest def _sub_raw_schema(data): """{block title: [subblock, …]} → dict · otherwise None (intermediate state of block B).""" if not isinstance(data, dict) or not data: return None out: dict[str, list[str]] = {} for k, v in data.items(): subs = _str_list(v) if isinstance(v, list) else None if not isinstance(k, str) or not k.strip() or not subs: return None out[k] = subs return out def _sidecar_schema(data): """{block title: [{title, level}, …]} → dict · otherwise None (sidecar with levels).""" if not isinstance(data, dict) or not data: return None for v in data.values(): if not isinstance(v, list) or not v: return None for s in v: if not isinstance(s, dict) or not str(s.get("title", "")).strip() or s.get("level") not in _LEVELS: return None return data def _relevance_complete(data) -> bool: """Does every subblock in the sidecar carry a valid relevance (relevant/peripheral)?""" if not isinstance(data, dict) or not data: return False return all( isinstance(s, dict) and s.get("relevance") in ("relevant", "peripheral") for v in data.values() if isinstance(v, list) for s in v ) def _question_pattern_chunk_schema(data) -> list[dict] | None: """{"pattern": [{block, subblock, question}, …]} → list of valid entries · otherwise None. One pattern per subblock (no type cross-product — the difficulty only comes at exam time from the learner's tier). Invalid individual entries are skipped.""" if not isinstance(data, dict) or not isinstance(data.get("pattern"), list): return None out = [] for e in data["pattern"]: if not isinstance(e, dict): continue blk = str(e.get("block", "")).strip() sub = str(e.get("subblock", "")).strip() question = str(e.get("question", "")).strip() if not blk or not sub or not question: continue out.append({"block": blk, "subblock": sub, "question": question}) return out or None def _question_pattern_complete(topic: str) -> bool: """Does the question-pattern sidecar exist (build ran)? Individual empty blocks fall back to live generation at exam time — so the file is enough.""" return isinstance(_json_file(question_pattern_path(topic)), dict) def _read(p: Path) -> str: return p.read_text(encoding="utf-8") if p.exists() else "" def _chunk_nums(items: list, n: int) -> list[list]: """Splits a flat list into n chunks as equal in size as possible.""" n = max(1, n) size = max(1, math.ceil(len(items) / n)) return [items[i:i + size] for i in range(0, len(items), size)] def _n_chunks(count: int, size: int = SUBBLOCK_CHUNK) -> int: return min(SUBBLOCK_MAX, max(1, math.ceil(count / size))) def _lpt_chunks(weights: list[int], target: int) -> list[list[int]]: """Distribute indices across chunks load-balanced (LPT, makespan-minimal). Weight = cost per index. K = ceil(total weight/target); heaviest first into the currently lightest bin. → index lists.""" if not weights: return [] K = max(1, math.ceil(sum(weights) / max(1, target))) bins: list[list[int]] = [[] for _ in range(K)] last = [0] * K for i in sorted(range(len(weights)), key=lambda x: weights[x], reverse=True): j = min(range(K), key=lambda b: last[b]) bins[j].append(i) last[j] += weights[i] return [b for b in bins if b] async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict | None: """Block B (DB + loop): per package, find subblocks in rounds (3 finders, until 0 new/cap), collect in the DB (≥2 mentions = consensus, 1× discarded), a judge cleans up per package. → {block title: [subblock, …]} (consensus) or None. Fills DB table `subblocks`.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] folder = source_folder(topic) caps = "files" if folder else "full" # Source for the evidence exam in the clarify step (discards invented/unsupportable subs). _type = load_source(topic).get("type", "thema") source = _prompt(_SOURCE_TEMPLATE[_type], project=folder) if _type in _SOURCE_TEMPLATE else _prompt("Blocks-Source-Thema", topic=topic) nums = list(entries) chunks = _chunk_nums(nums, _n_chunks(len(nums))) n = len(chunks) title_by_num = {num: _title(entries[num]) for num in nums} norm_by_num = {num: _norm_title(title_by_num[num]) for num in nums} await db.delete_subblocks(topic) # fresh start of the block (idempotent counter) async def _known_block(chunk): known = [] for num in chunk: subs = [s["sub_title"] for s in await db.list_subblocks(topic, norm_by_num[num])] if subs: known.append(f"\n" + "\n".join(f"- {s}" for s in subs)) if not known: return "" # Do NOT list known items again (otherwise re-confirmation inflates the mention count, # self-bias/echo) — only add what's missing. This keeps the counter an honest consensus signal. return ("\n\nBEREITS ERFASST — liste diese NICHT erneut. Finde nur, was FEHLT:\n" + "\n".join(known)) # Phase "Subblocks find": per package loop until 0 new subs / time cap. async def _find(c, chunk): assignment = "\n".join(f"- {entries[num]}" for num in chunk) chunk_idx = _title_index({num: title_by_num[num] for num in chunk}) start = time.monotonic() round_n = 0 while not is_cancelled(): round_n += 1 bekannt = await _known_block(chunk) if round_n > 1 else "" paths = [work_dir / f"subblock-c{c}-r{round_n}-{i}.md" for i in (1, 2, 3)] for p in paths: p.unlink(missing_ok=True) slots = [{ "key": f"blocks-{topic}-subblock-c{c}-r{round_n}-{i}", "prompt": _prompt("Subblock-Research", topic=topic, assignment=assignment, known=bekannt, out_path=p, extra=_extra(instructions)), "role": "quick", "capabilities": caps, "payload": (lambda result, p=p: _parse_subblocks(_read(p)) or None), } for i, p in enumerate(paths, 1)] agent_texts = await _race(topic, f"Subblocks package {c} R{round_n}", slots, 2, _timeout("subblock", len(chunk)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) if is_cancelled(): return False if not agent_texts: return round_n > 1 # round 1 without result = error; later = simply the end existing = {num: {s["sub_norm"] for s in await db.list_subblocks(topic, norm_by_num[num])} for num in chunk} new = 0 for d in agent_texts: for marker, subs in d.items(): num = _resolve_title(chunk_idx, marker) if num is None: continue seen_set = set() for sub in subs: sn = _norm_title(sub) if not sn or sn in seen_set: continue seen_set.add(sn) if sn not in existing[num]: new += 1 existing[num].add(sn) await db.upsert_subblock(topic, norm_by_num[num], sn, title_by_num[num], sub) if new == 0: break if time.monotonic() - start > SUBBLOCK_CAP: _log(topic, f"Subblocks package {c}: time cap reached (round {round_n})") break return True oks = await _gather_progress([_find(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _report_p(set_p, topic, "Subblocks find")) if is_cancelled(): return None if not all(ok is True for ok in oks): _blocks_errors[topic] = "Subblocks failed (research)" return None # Phase "Subblocks select": ≥2 mentions = consensus, 1× discarded (code). set_p(f"Subblocks select ({n} packages)…", step=_step_idx(topic, "Subblocks select")) for num in nums: for s in await db.list_subblocks(topic, norm_by_num[num]): await db.set_subblock_fields(topic, norm_by_num[num], s["sub_norm"], status=("consensus" if s["mentions"] >= 2 else "discarded")) # Phase "Subblocks clarify": source panel (SUBBAUSTEIN_PANEL judges) checks consensus + uncertain (1×) # against the source; code majority per sub. External, multi-voice gate against single-judge bias + echo. async def _clarify(c, chunk): fp = work_dir / f"subblock-final-c{c}.md" if _parse_subblocks(_read(fp)): return block_texts, has_any = [], False consensus_by_num: dict[int, list[str]] = {} for num in chunk: rows = await db.list_subblocks(topic, norm_by_num[num]) consensus_subs = [s["sub_title"] for s in rows if s["status"] == "consensus"] uncertain = [s["sub_title"] for s in rows if s["status"] != "consensus" and s["mentions"] == 1] consensus_by_num[num] = consensus_subs if not consensus_subs and not uncertain: continue has_any = True k_lines = "\n".join(f"- {s}" for s in consensus_subs) if consensus_subs else "- (keiner)" u_lines = "\n".join(f"- {s}" for s in uncertain) if uncertain else "- (keiner)" block_texts.append(f"BLOCK: {title_by_num[num]}\nKonsens (≥2 finders):\n{k_lines}\nUnsicher (1× — streng gegen Source check):\n{u_lines}") if not has_any: return chunk_idx = _title_index({num: title_by_num[num] for num in chunk}) paths = [work_dir / f"subblock-final-c{c}-j{j}.md" for j in range(1, SUBBLOCK_PANEL + 1)] pending = [(j, p) for j, p in enumerate(paths, 1) if _parse_subblocks(_read(p)) is None] for _, p in pending: p.unlink(missing_ok=True) if pending: slots = [{ "key": f"blocks-{topic}-subblock-final-c{c}-j{j}", "prompt": _prompt("Subblock-Mapping", topic=topic, source=source, blocks="\n\n".join(block_texts), out_path=p, extra=_extra(instructions)), "role": "judge", "capabilities": caps, "payload": (lambda result, p=p: _parse_subblocks(_read(p)) or None), } for j, p in pending] existing = SUBBLOCK_PANEL - len(pending) await _race(topic, f"Subblock-Clarification {c}", slots, max(1, 2 - existing), _timeout("subblock_check", len(chunk)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) if is_cancelled(): return outs = [d for p in paths if (d := _parse_subblocks(_read(p)))] if not outs: # panel fully failed → adopt consensus (the fallback as before) _log(topic, f"Subblock clarification package {c} failed — consensus adopted") text = "\n\n".join(f"\n" + "\n".join(f"- {s}" for s in consensus_by_num[num]) for num in chunk if consensus_by_num[num]) atomic_write_text(fp, text) return # code majority per block/sub-norm: keep if a majority of judges list it (tie → keep). block_texts_out = [] for num in chunk: votes: dict[str, int] = {} form: dict[str, str] = {} for d in outs: seen = set() for marker, subs in d.items(): if _resolve_title(chunk_idx, marker) != num: continue for sub in subs: sn = _norm_title(sub) if not sn or sn in seen: continue seen.add(sn) form.setdefault(sn, sub) votes[sn] = votes.get(sn, 0) + 1 kept = [form[sn] for sn in form if votes[sn] * 2 >= len(outs)] if kept: block_texts_out.append(f"\n" + "\n".join(f"- {s}" for s in kept)) atomic_write_text(fp, "\n\n".join(block_texts_out)) await _gather_progress([_clarify(c, chunk) for c, chunk in enumerate(chunks, 1)], n, _report_p(set_p, topic, "Subblocks clarify")) if is_cancelled(): return None # Final list per block: judge output, otherwise consensus fallback. Reconcile DB + build raw. raw: dict[str, list[str]] = {} for c, chunk in enumerate(chunks, 1): final = _parse_subblocks(_read(work_dir / f"subblock-final-c{c}.md")) or {} chunk_idx = _title_index({num: title_by_num[num] for num in chunk}) final_by_num = {_resolve_title(chunk_idx, m): subs for m, subs in final.items() if _resolve_title(chunk_idx, m) is not None} for num in chunk: title = title_by_num[num] consensus = [s["sub_title"] for s in await db.list_subblocks(topic, norm_by_num[num]) if s["status"] == "consensus"] subs = final_by_num.get(num) or consensus if not subs: continue raw[title] = subs # align DB to the final list: final = consensus, rest discarded, add new ones. final_norms = {_norm_title(s) for s in subs} have = {s["sub_norm"] for s in await db.list_subblocks(topic, norm_by_num[num])} for s in await db.list_subblocks(topic, norm_by_num[num]): await db.set_subblock_fields(topic, norm_by_num[num], s["sub_norm"], status=("consensus" if s["sub_norm"] in final_norms else "discarded")) for s in subs: sn = _norm_title(s) if sn and sn not in have: await db.upsert_subblock(topic, norm_by_num[num], sn, title, s) await db.set_subblock_fields(topic, norm_by_num[num], sn, status="consensus") await _dedup_subblocks(topic, raw) # near-dup filter per block (deterministic, no LLM) if not raw: _blocks_errors[topic] = "No subblocks determined" return None return raw async def _dedup_subblocks(topic: str, raw: dict[str, list[str]]) -> None: """Deterministic near-duplicate filter per block: subblocks with cosine ≥ EMBEDDING_SUB_DUP are the same statement (reliable in the narrow block context — no LLM needed). Per duplicate group keeps the most informative (longest); rest → DB discarded + out of `roh`. Model missing → silently skip (like the rest of the embedding fallback).""" if not EMBEDDING_AKTIV or not await asyncio.to_thread(embedding.available): return for title, subs in list(raw.items()): if len(subs) < 2: continue sims = await asyncio.to_thread(embedding.embed_sims, subs) if sims is None: return keepers: list[int] = [] discarded: list[int] = [] for i in sorted(range(len(subs)), key=lambda x: (-len(subs[x]), x)): # most informative first if any(float(sims[i][j]) >= EMBEDDING_SUB_DUP for j in keepers): discarded.append(i) else: keepers.append(i) if not discarded: continue bnorm = _norm_title(title) for i in discarded: await db.set_subblock_fields(topic, bnorm, _norm_title(subs[i]), status="discarded") raw[title] = [subs[i] for i in sorted(keepers)] # original order of the kept ones def _code_vote(rater: list[dict], n: int) -> tuple[dict, dict]: """Majority vote over rater dicts on local ids 1..n → (outcome, disputed). A clear winner needs ≥2 votes and no tie; otherwise the id is disputed (kept with its vote list).""" outcome: dict[int, str] = {} disputed: dict[int, list[str]] = {} for k in range(1, n + 1): vote_list = [d[k] for d in rater if k in d] counter: dict[str, int] = {} for s in vote_list: counter[s] = counter.get(s, 0) + 1 best = max(counter.values(), default=0) winners = [s for s, v in counter.items() if v == best] if len(winners) == 1 and best >= 2: outcome[k] = winners[0] else: disputed[k] = vote_list return outcome, disputed def _disputed_lines(items, item_idxs, disputed: dict) -> str: """Render disputed items as `k. [block] sub — Stimmen: a, b` lines for the judge prompt.""" return "\n".join( f"{k}. [{items[item_idxs[k - 1]][0]}] {items[item_idxs[k - 1]][1]} — Stimmen: {', '.join(vote_list) or 'none'}" for k, vote_list in disputed.items() ) async def _levels_block(ctx: GenContext, set_p, files: dict, raw: dict, instructions: str) -> dict | None: """Block C: three phases with a barrier — find (classify), select (vote), clarify. Local IDs 1..n per package, mapped to global gid afterwards. → {block title: [{title, level}, …]} or None.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] # key points per sub as concise context (better-grounded classification; classification needs little). facts_map = _json_file(files["facts"]) facts_map = facts_map if isinstance(facts_map, dict) else {} items = [(title, sub) for title, subs in raw.items() for sub in subs] # global id = index+1 if not items: return {title: [] for title in raw} # pack chunks from WHOLE blocks (don't split a block) → each rater sees per block # all subs and can classify relatively. Item indices per block in raw order. chunks, cur, i = [], [], 0 for _title_b, subs in raw.items(): g = list(range(i, i + len(subs))) i += len(subs) if cur and len(cur) + len(g) > LEVEL_CHUNK: chunks.append(cur) cur = [] cur.extend(g) if cur: chunks.append(cur) n = len(chunks) def rater_paths(c): return [work_dir / f"level-c{c}-{i}.json" for i in (1, 2, 3)] def lset(item_idxs): return set(range(1, len(item_idxs) + 1)) # Phase "Levels find": 3 raters per package (min. 2), local IDs. async def _rate(c, item_idxs): local_set = lset(item_idxs) paths = rater_paths(c) existing = sum(1 for p in paths if _levels_schema(_json_file(p), local_set)) if existing >= 2: return True enum_lines, cur_b = [], None for k, j in enumerate(item_idxs, 1): b, sub = items[j] if b != cur_b: enum_lines.append(f"\nBAUSTEIN: {b}") cur_b = b enum_lines.append(f"{k}. {sub}") if (kz := _core_line(facts_map.get(b, {}).get(_norm_title(sub)))): enum_lines.append(f" {kz}") enum = "\n".join(enum_lines).strip() pending = [(i, p) for i, p in enumerate(paths, 1) if not _levels_schema(_json_file(p), local_set)] slots = [{ "key": f"blocks-{topic}-level-c{c}-{i}", "prompt": _prompt("Levels-Research", topic=topic, subblocks=enum, out_path=p, extra=_extra(instructions)), "role": "fast", "capabilities": "files", "payload": (lambda result, p=p, ids=local_set: _levels_schema(_json_file(p), ids)), } for i, p in pending] new = await _race(topic, f"Levels package {c}", slots, 2 - existing, _timeout("level", len(item_idxs)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) return not is_cancelled() and new is not None oks = await _gather_progress([_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _report_p(set_p, topic, "Levels find")) if is_cancelled(): return None if not all(ok is True for ok in oks): _blocks_errors[topic] = "Classification failed (research)" return None # Phase "Levels select": code vote per package → (outcome, disputed). set_p(f"Levels select ({n} packages)…", step=_step_idx(topic, "Levels select")) vote_by_c = {} for c, item_idxs in enumerate(chunks, 1): local_set = lset(item_idxs) rater = [d for p in rater_paths(c) if (d := _levels_schema(_json_file(p), local_set))] vote_by_c[c] = _code_vote(rater, len(item_idxs)) # Phase "Levels clarify": one judge per package on the disputed items, all in parallel. async def _clarify(c, item_idxs): outcome, strittig = vote_by_c[c] if strittig: judge_path = work_dir / f"level-final-c{c}.json" decision = _levels_schema(_json_file(judge_path), set(strittig)) if decision is None: disputed_block = _disputed_lines(items, item_idxs, strittig) status, decision = await run_single_slot( ctx, f"Levels-Clarification {c}", key=f"blocks-{topic}-level-final-c{c}", prompt=_prompt("Levels-Mapping", topic=topic, disputed=disputed_block, out_path=judge_path, extra=_extra(instructions)), role="judge", capabilities="files", payload=lambda result, p=judge_path, ids=set(strittig): _levels_schema(_json_file(p), ids), timeout=_timeout("level_check", len(strittig)), ) if status == FAILED: _log(topic, f"Levels clarification package {c} failed — default 'advanced'") decision = decision if isinstance(decision, dict) else {} # disputed without a decision → 'advanced'; vote winners stay; judge overrides. outcome = {**{k: "advanced" for k in strittig}, **outcome, **decision} return {item_idxs[k - 1] + 1: level for k, level in outcome.items()} parts = await _gather_progress([_clarify(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _report_p(set_p, topic, "Levels clarify")) if is_cancelled(): return None level_by_id: dict[int, str] = {} for c, part in enumerate(parts, 1): if not isinstance(part, dict): # clarification is not fatal: vote outcome + default 'advanced' for disputed. if isinstance(part, BaseException): _log(topic, f"Levels clarification package {c}: {type(part).__name__}: {part}") outcome, strittig = vote_by_c[c] item_idxs = chunks[c - 1] merged = {**{k: "advanced" for k in strittig}, **outcome} part = {item_idxs[k - 1] + 1: s for k, s in merged.items()} level_by_id.update(part) # assemble the sidecar — same order as items → gid matches sidecar: dict[str, list[dict]] = {} gid = 0 for title, subs in raw.items(): lst = [] for sub in subs: gid += 1 lst.append({"title": sub, "level": level_by_id.get(gid, "advanced")}) sidecar[title] = lst return sidecar _FACTS_FIELDS = ("key_points", "prerequisites", "hurdles", "cited_facts", "example_idea") def _facts_schema(data) -> list[dict] | None: """{"facts": [{block, subblock, …}]} → valid list · otherwise None. Strictly separates belegte_facts (with source) from example_idee (generative).""" if not isinstance(data, dict) or not isinstance(data.get("facts"), list): return None out = [] for e in data["facts"]: if not isinstance(e, dict): continue blk = str(e.get("block", "")).strip() sub = str(e.get("subblock", "")).strip() if not blk or not sub: continue bf = [{"text": t, "source": str(f.get("source", "")).strip()} for f in (e.get("cited_facts") or []) if isinstance(f, dict) and (t := str(f.get("text", "")).strip())] out.append({ "block": blk, "subblock": sub, "key_points": [k for x in (e.get("key_points") or []) if (k := str(x).strip())], "prerequisites": str(e.get("prerequisites", "")).strip(), "hurdles": str(e.get("hurdles", "")).strip(), "cited_facts": bf, "example_idea": str(e.get("example_idea", "")).strip(), }) return out or None def _facts_check_schema(data) -> list[tuple[str, bool]] | None: """Facts check → [(sub_norm, verwerfen)] per objection · {ok:true}→[] · None if invalid. verwerfen=True: sub not supportable in substance (remove). verwerfen=False: only correct the fact.""" if not isinstance(data, dict): return None if data.get("ok") is True: return [] pr = data.get("problems") if not isinstance(pr, list): return None return [(sn, bool(p.get("discard"))) for p in pr if isinstance(p, dict) and (sn := _norm_title(str(p.get("subblock", ""))))] def _core_line(fk) -> str: """Concise key-point line for classification (level/relevance) — less context suffices there. Empty if no facts/key points (legacy).""" if not isinstance(fk, dict) or not fk.get("key_points"): return "" return "Kern: " + " · ".join(str(k) for k in fk["key_points"]) def _facts_lines(fk: dict) -> str: z = [] if fk.get("key_points"): z.append("Kernpunkte: " + " · ".join(str(k) for k in fk["key_points"])) if fk.get("prerequisites"): z.append("Voraussetzung: " + fk["prerequisites"]) if fk.get("hurdles"): z.append("Hürde: " + fk["hurdles"]) for bf in fk.get("cited_facts", []): z.append(f"FAKT: {bf['text']} (Source: {bf.get('source', '?')})") if fk.get("example_idea"): z.append("Example: " + fk["example_idea"]) return "\n".join(z) def _facts_complete(files: dict) -> bool: """Does the facts map exist (block done)? {block: {sub_norm: {...}}}.""" d = _json_file(files["facts"]) return isinstance(d, dict) and bool(d) async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, instructions: str) -> tuple | None: """Block: per sub extract source facts (find) → verify (check) → correct/discard (fix). Extract-once grounding: the result feeds level/relevance/questions/guide. → (facts_map, discarded_map) — facts_map {block: {sub_norm: facts}}, discarded_map {block: {sub_norm}} (unsupportable subs to remove) — or None on cancel/error.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] caps = "files" if folder else "full" type = q.get("type", "thema") source = _prompt(_SOURCE_TEMPLATE[type], project=folder) if type in _SOURCE_TEMPLATE else _prompt("Blocks-Source-Thema", topic=topic) blocks = [(title, [str(s).strip() for s in subs if str(s).strip()]) for title, subs in raw.items() if subs] if not blocks: return {}, {} chunks = _lpt_chunks([len(subs) for _, subs in blocks], FACTS_CHUNK_SUBS) def raw_path(ci): return work_dir / f"facts-c{ci}.json" def supp_path(ci): return work_dir / f"facts-erg-c{ci}.json" def chk_path(ci, j): return work_dir / f"facts-check-c{ci}-j{j}.json" def fix_path(ci): return work_dir / f"facts-fix-c{ci}.json" def ctitle(idxs): return [blocks[i][0] for i in idxs] def block_text(idxs): return "\n\n".join( f"BLOCK: {blocks[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in blocks[i][1]) for i in idxs) # raw facts of a chunk → {block: {sub_norm: {sub, …fields}}}, matched to chunk titles. def raw_map(ci, path): idxs = chunks[ci] rel_by = {blocks[i][0]: blocks[i][1] for i in idxs} ct = ctitle(idxs) out: dict[str, dict] = {} for e in _facts_schema(_json_file(path)) or []: bt = _match_sub(e["block"], ct) if bt not in rel_by: continue sub = _match_sub(e["subblock"], rel_by[bt]) out.setdefault(bt, {})[_norm_title(sub)] = {"sub": sub, **{k: e[k] for k in _FACTS_FIELDS}} return out # union raw facts + completeness supplements (recall): only subs that exist in raw. def _chunk_facts(ci): raw = raw_map(ci, raw_path(ci)) erg = raw_map(ci, supp_path(ci)) if supp_path(ci).exists() else {} if not erg: return raw for bt, fm in raw.items(): ebt = erg.get(bt, {}) for sn, fk in fm.items(): ek = ebt.get(sn) if not ek: continue seen = {str(k).strip().casefold() for k in fk.get("key_points", [])} for k in ek.get("key_points", []): if str(k).strip().casefold() not in seen: seen.add(str(k).strip().casefold()) fk["key_points"].append(k) seent = {bf["text"].strip().casefold() for bf in fk.get("cited_facts", [])} for bf in ek.get("cited_facts", []): if bf["text"].strip().casefold() not in seent: seent.add(bf["text"].strip().casefold()) fk["cited_facts"].append(bf) for f in ("prerequisites", "hurdles", "example_idea"): if not fk.get(f) and ek.get(f): fk[f] = ek[f] return raw # 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) status, _r = await run_single_slot( ctx, f"Facts {ci}", key=f"blocks-{topic}-facts-c{ci}", prompt=_prompt("Facts-Research", topic=topic, source=source, blocks=block_text(idxs), out_path=fp, extra=_extra(instructions)), role="guide", capabilities=caps, payload=lambda result, p=fp: _facts_schema(_json_file(p)), timeout=_timeout("content", subs_total)) return status != FAILED and _facts_schema(_json_file(fp)) is not None oks = await _gather_progress([_find(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _report_p(set_p, topic, "Facts find")) if is_cancelled(): return None if not any(ok is True for ok in oks): _blocks_errors[topic] = "Facts extraction failed" return None # Phase "Facts ergänzen" (recall): a targeted gap hunt per chunk looks for source-backed facts that # the single pass missed. Best-effort — never fails (no erg file → merge uses only raw). async def _supplement(ci, idxs): ep = supp_path(ci) if _facts_schema(_json_file(ep)): return per = raw_map(ci, raw_path(ci)) if not per: return block = "\n\n".join( f"BLOCK: {bt}\nSUBBAUSTEINE (mit bereits erfassten Facts):\n" + "\n".join( f"- {fk['sub']}\n Erfasst: " + ("; ".join( list(fk.get("key_points", [])) + [bf["text"] for bf in fk.get("cited_facts", [])]) or "(nichts)") for fk in fm.values()) for bt, fm in per.items()) subs_total = sum(len(blocks[i][1]) for i in idxs) await run_single_slot( ctx, f"Facts supplement {ci}", key=f"blocks-{topic}-facts-erg-c{ci}", prompt=_prompt("Facts-Supplement", topic=topic, source=source, blocks=block, out_path=ep, extra=_extra(instructions)), role="guide", capabilities=caps, payload=lambda result, p=ep: _facts_schema(_json_file(p)), timeout=_timeout("content", subs_total)) set_p("Facts supplement…", step=_step_idx(topic, "Facts find")) await _gather_progress([_supplement(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _report_p(set_p, topic, "Facts find")) if is_cancelled(): return None # Phase "Facts check": FACTS_CHECK_PANEL judges per chunk. Two majority sets: # flagged (fact inaccurate → correct) and discard (sub not supportable → remove). async def _check(ci, idxs): per = _chunk_facts(ci) # raw + supplements → panel verifies the union if not per: return ci, set(), set() facts_text = "\n\n".join(f"SUBBAUSTEIN: {fk['sub']}\n{_facts_lines(fk)}" for fm in per.values() for fk in fm.values()) pending = [j for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if _facts_check_schema(_json_file(chk_path(ci, j))) is None] await asyncio.gather(*[ run_agent(f"blocks-{topic}-facts-check-c{ci}-j{j}", _prompt("Facts-Check", topic=topic, source=source, facts=facts_text, out_path=chk_path(ci, j), extra=_extra(instructions)), _timeout("content_check", len(per)), provider=provider, role="judge", capabilities=caps) for j in pending], return_exceptions=True) outs = [s for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if (s := _facts_check_schema(_json_file(chk_path(ci, j)))) is not None] bvotes: dict[str, int] = {} vvotes: dict[str, int] = {} for s in outs: # s = [(sub_norm, verwerfen)] of one judge gb, gv = set(), set() for sn, disc in s: if sn not in gb: gb.add(sn); bvotes[sn] = bvotes.get(sn, 0) + 1 if disc and sn not in gv: gv.add(sn); vvotes[sn] = vvotes.get(sn, 0) + 1 threshold = len(outs) / 2 if outs else 99 flagged = {sn for sn, v in bvotes.items() if v > threshold} # Discarding is irreversible → stricter than flagging: majority AND ≥2 agreeing judges # (prevents deletion by a single vote when the panel is degraded). to_discard = {sn for sn, v in vvotes.items() if v > threshold and v >= 2} return ci, flagged, to_discard check = await _gather_progress([_check(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _report_p(set_p, topic, "Facts check")) if is_cancelled(): return None flagged: dict[int, set] = {} to_discard: dict[int, set] = {} for r in check: if isinstance(r, tuple) and len(r) == 3: ci, b, v = r flagged[ci] = b to_discard[ci] = v # Phase "Facts fix": re-extract only CORRECTABLE ones (flagged without discard). correctable = {ci: (flagged.get(ci, set()) - to_discard.get(ci, set())) for ci in flagged} n_problem = sum(len(s) for s in correctable.values()) if n_problem: set_p(f"Correcting facts ({n_problem})…", step=_step_idx(topic, "Facts fix")) async def _fix(ci): subs_norm = correctable.get(ci, set()) if not subs_norm or _facts_schema(_json_file(fix_path(ci))): return idxs = chunks[ci] rel_by = {blocks[i][0]: blocks[i][1] for i in idxs} goal = [] for bt, subs in rel_by.items(): affected_subs = [s for s in subs if _norm_title(s) in subs_norm] if affected_subs: goal.append(f"BLOCK: {bt}\nSUBBAUSTEINE:\n" + "\n".join(f"- {s}" for s in affected_subs)) if not goal: return await run_single_slot( ctx, f"Facts-Fix {ci}", key=f"blocks-{topic}-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="guide", capabilities=caps, payload=lambda result, p=fix_path(ci): _facts_schema(_json_file(p)), timeout=_timeout("content", len(subs_norm))) await _gather_progress([_fix(ci) for ci in correctable], len(correctable), _report_p(set_p, topic, "Facts fix")) if is_cancelled(): return None # assemble: raw + fix overrides for corrected. Discarded subs out (+ report per block). outcome: dict[str, dict] = {} discarded_map: dict[str, set] = {} for ci in range(len(chunks)): per = _chunk_facts(ci) # raw + supplements (recall); fix overrides only corrected fix = raw_map(ci, fix_path(ci)) if fix_path(ci).exists() else {} disc = to_discard.get(ci, set()) for bt, fm in per.items(): for sn, fk in fm.items(): if sn in disc: discarded_map.setdefault(bt, set()).add(sn) continue winners = fix.get(bt, {}).get(sn, fk) if sn in correctable.get(ci, set()) else fk outcome.setdefault(bt, {})[sn] = {k: winners[k] for k in _FACTS_FIELDS} if discarded_map: _log(topic, f"Facts check discards {sum(len(s) for s in discarded_map.values())} unsupportable subblocks") return outcome, discarded_map async def _relevance_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None: """Block D: three phases with a barrier — find (relevant/peripheral), select (vote), clarify. Items from the sidecar; local IDs 1..n per package → global gid. → {gid: relevance} or None on cancel/research error. Default on gap/dispute: 'relevant'.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] items = [(title, sub["title"], sub.get("facts")) for title, subs in sidecar.items() for sub in subs] # global id = index+1 if not items: return {} chunks = _chunk_nums(list(range(len(items))), _n_chunks(len(items), LEVEL_CHUNK)) n = len(chunks) def rater_paths(c): return [work_dir / f"relevance-c{c}-{i}.json" for i in (1, 2, 3)] def lset(item_idxs): return set(range(1, len(item_idxs) + 1)) # Phase "Relevance find": 3 raters per package (min. 2), local IDs. async def _rate(c, item_idxs): local_set = lset(item_idxs) paths = rater_paths(c) existing = sum(1 for p in paths if _relevance_schema(_json_file(p), local_set)) if existing >= 2: return True enum_lines = [] for k, j in enumerate(item_idxs, 1): enum_lines.append(f"{k}. [{items[j][0]}] {items[j][1]}") if (kz := _core_line(items[j][2])): enum_lines.append(f" {kz}") enum = "\n".join(enum_lines) pending = [(i, p) for i, p in enumerate(paths, 1) if not _relevance_schema(_json_file(p), local_set)] slots = [{ "key": f"blocks-{topic}-relevance-c{c}-{i}", "prompt": _prompt("Relevance-Research", topic=topic, subblocks=enum, out_path=p, extra=_extra(instructions)), "role": "fast", "capabilities": "files", "payload": (lambda result, p=p, ids=local_set: _relevance_schema(_json_file(p), ids)), } for i, p in pending] new = await _race(topic, f"Relevance package {c}", slots, 2 - existing, _timeout("relevance", len(item_idxs)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) return not is_cancelled() and new is not None oks = await _gather_progress([_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _report_p(set_p, topic, "Relevance find")) if is_cancelled(): return None if not all(ok is True for ok in oks): _blocks_errors[topic] = "Relevance failed (research)" return None # Phase "Relevance select": code vote per package → (outcome, disputed). set_p(f"Relevance select ({n} packages)…", step=_step_idx(topic, "Relevance select")) vote_by_c = {} for c, item_idxs in enumerate(chunks, 1): local_set = lset(item_idxs) rater = [d for p in rater_paths(c) if (d := _relevance_schema(_json_file(p), local_set))] vote_by_c[c] = _code_vote(rater, len(item_idxs)) # Phase "Relevance clarify": one judge per package on the disputed items, all in parallel. async def _clarify(c, item_idxs): outcome, strittig = vote_by_c[c] if strittig: judge_path = work_dir / f"relevance-final-c{c}.json" decision = _relevance_schema(_json_file(judge_path), set(strittig)) if decision is None: disputed_block = _disputed_lines(items, item_idxs, strittig) status, decision = await run_single_slot( ctx, f"Relevance-Clarification {c}", key=f"blocks-{topic}-relevance-final-c{c}", prompt=_prompt("Relevance-Mapping", topic=topic, disputed=disputed_block, out_path=judge_path, extra=_extra(instructions)), role="judge", capabilities="files", payload=lambda result, p=judge_path, ids=set(strittig): _relevance_schema(_json_file(p), ids), timeout=_timeout("relevance_check", len(strittig)), ) if status == FAILED: _log(topic, f"Relevance clarification package {c} failed — default 'relevant'") decision = decision if isinstance(decision, dict) else {} # disputed without a decision → 'relevant' (never accidentally exclude). outcome = {**{k: "relevant" for k in strittig}, **outcome, **decision} return {item_idxs[k - 1] + 1: rel for k, rel in outcome.items()} parts = await _gather_progress([_clarify(c, idxs) for c, idxs in enumerate(chunks, 1)], len(chunks), _report_p(set_p, topic, "Relevance clarify")) if is_cancelled(): return None relevance_by_id: dict[int, str] = {} for c, part in enumerate(parts, 1): if not isinstance(part, dict): # clarification is not fatal: vote outcome + default 'relevant' for disputed. if isinstance(part, BaseException): _log(topic, f"Relevance clarification package {c}: {type(part).__name__}: {part}") outcome, strittig = vote_by_c[c] item_idxs = chunks[c - 1] merged = {**{k: "relevant" for k in strittig}, **outcome} part = {item_idxs[k - 1] + 1: s for k, s in merged.items()} relevance_by_id.update(part) return relevance_by_id def _match_sub(agent_sub: str, rel: list[str]) -> str: """Map the agent's subblock title to the matching relevant title — exact, then normalized, then substring (the agent drops e.g. the prefix "Question: "). No match → keep the agent title. This way NO pattern is lost to a title mismatch.""" if agent_sub in rel: return agent_sub an = _norm_title(agent_sub) for r in rel: rn = _norm_title(r) if an and rn and (an == rn or an in rn or rn in an): return r return agent_sub async def _question_pattern_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None: """Block E (chunks of 10): find (1 generator per ~10 blocks, parallel), select (code: group per block + dedup), clarify (1 critic per chunk), check (catch-up round). Assignment per entry via the `block` field (a chunk file carries several blocks). → {block title: [{subblock, question}, …]} or None on cancel.""" topic, is_cancelled = ctx.topic, ctx.is_cancelled work_dir = files["arbeit"] # ALL subblocks (including peripheral) get a pattern — peripheral is testable in the FGuide level. # facts_by: full facts context per sub (generation benefits — better questions). blocks = [] facts_by: dict[tuple, dict] = {} for title, subs in sidecar.items(): all_titles = [] for s in subs: if isinstance(s, dict) and (st := str(s.get("title", "")).strip()): all_titles.append(st) if isinstance(s.get("facts"), dict): facts_by[(title, _norm_title(st))] = s["facts"] if all_titles: blocks.append((title, all_titles)) if not blocks: return {} chunks = _lpt_chunks([len(rel) for _, rel in blocks], QUESTION_CHUNK_SUBS) # load-balanced by sub count def raw_path(ci): return work_dir / f"question-pattern-c{ci}.json" def final_path(ci): return work_dir / f"question-pattern-final-c{ci}.json" def _chunk_title(idxs): return [blocks[i][0] for i in idxs] # Phase "Questions find": 1 generator per chunk, all in parallel. async def _find(ci, idxs): fp = raw_path(ci) if _question_pattern_chunk_schema(_json_file(fp)): return # Resume def _sub_line(bi, s): line = f"- {s}" fk = facts_by.get((blocks[bi][0], _norm_title(s))) if fk and (ft := _facts_lines(fk)): line += "\n" + "\n".join(" " + l for l in ft.split("\n")) return line block = "\n\n".join( f"BLOCK: {blocks[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(_sub_line(i, s) for s in blocks[i][1]) for i in idxs ) subs_total = sum(len(blocks[i][1]) for i in idxs) status, _ = await run_single_slot( ctx, f"Question-Pattern {ci}", key=f"blocks-{topic}-question-pattern-c{ci}", prompt=_prompt("Question-Pattern-Research", topic=topic, blocks=block, out_path=fp, extra=_extra(instructions)), role="fast", capabilities="files", payload=lambda result, p=fp: _question_pattern_chunk_schema(_json_file(p)), timeout=_timeout("question_pattern", subs_total), ) if status == FAILED: _log(topic, f"Question pattern chunk {ci} failed — blocks in fallback (catch-up round/live)") async def find_all(ci_list): ci_list = list(ci_list) await _gather_progress([_find(ci, chunks[ci]) for ci in ci_list], len(ci_list), _report_p(set_p, topic, "Questions find")) await find_all(range(len(chunks))) if is_cancelled(): return None # Phase "Questions select": code — group chunk files per block, drop duplicates, # loosely map block/subblock titles to the targets (discard nothing for a mismatch). def _select_chunk(ci): idxs = chunks[ci] ctitle = _chunk_title(idxs) rel_by = {blocks[i][0]: blocks[i][1] for i in idxs} out, seen_set = {}, {} for e in _question_pattern_chunk_schema(_json_file(raw_path(ci))) or []: title = _match_sub(e["block"], ctitle) if title not in rel_by: continue # not assignable → discard sub = _match_sub(e["subblock"], rel_by[title]) seen = seen_set.setdefault(title, set()) if sub in seen: continue # exactly one pattern per subblock seen.add(sub) out.setdefault(title, []).append({"subblock": sub, "question": e["question"]}) return out def _select_all(ci_list): raw = {} for ci in ci_list: for title, eintraege in _select_chunk(ci).items(): raw.setdefault(title, []).extend(eintraege) return raw set_p("Questions select…", step=_step_idx(topic, "Questions select")) raw_by_title = _select_all(range(len(chunks))) # Phase "Questions clarify": 1 critic per chunk cleans up the tables (grouped by block). async def _clarify(ci, idxs): fp = final_path(ci) if _question_pattern_chunk_schema(_json_file(fp)): return # resume block_texts = [] for i in idxs: t = blocks[i][0] eintraege = raw_by_title.get(t) or [] if not eintraege: continue lines = "\n".join(f"- ({e['subblock']}) {e['question']}" for e in eintraege) block_texts.append(f"BLOCK: {t}\n{lines}") if not block_texts: return # nothing to clarify in this chunk subs_total = sum(len(blocks[i][1]) for i in idxs) status, _ = await run_single_slot( ctx, f"Question-Pattern-Clarification {ci}", key=f"blocks-{topic}-question-pattern-final-c{ci}", prompt=_prompt("Question-Pattern-Critique", topic=topic, table="\n\n".join(block_texts), out_path=fp, extra=_extra(instructions)), role="judge", capabilities="files", payload=lambda result, p=fp: _question_pattern_chunk_schema(_json_file(p)), timeout=_timeout("question_pattern_check", subs_total), ) if status == FAILED: _log(topic, f"Question pattern clarification chunk {ci} failed — raw pattern adopted") async def clarify_all(ci_list): ci_list = list(ci_list) await _gather_progress([_clarify(ci, chunks[ci]) for ci in ci_list], len(ci_list), _report_p(set_p, topic, "Questions clarify")) await clarify_all(range(len(chunks))) if is_cancelled(): return None # Clarified chunk table per block, fallback to raw pattern. Map titles loosely. def _final_by_title(ci_list): out = {} for ci in ci_list: idxs = chunks[ci] ctitle = _chunk_title(idxs) rel_by = {blocks[i][0]: blocks[i][1] for i in idxs} for e in _question_pattern_chunk_schema(_json_file(final_path(ci))) or []: title = _match_sub(e["block"], ctitle) if title not in rel_by: continue out.setdefault(title, []).append( {"subblock": _match_sub(e["subblock"], rel_by[title]), "question": e["question"]}) return out final_by_title = _final_by_title(range(len(chunks))) outcome = {t: (final_by_title.get(t) or raw_by_title.get(t) or []) for t, _ in blocks} # Phase "Questions check": per-sub completeness. Generators crash randomly (~15 %), # 1 agent per chunk without retry → subs (whole blocks) fall through silently. Hence several # rounds that re-request ONLY the missing subs (short packages, Question-Pattern-Research). set_p("Questions check…", step=_step_idx(topic, "Questions check")) def _missing_subs() -> list[tuple[str, list[str]]]: out = [] for t, subs in blocks: have_set = {_norm_title(e["subblock"]) for e in outcome.get(t) or []} miss = [s for s in subs if _norm_title(s) not in have_set] if miss: out.append((t, miss)) return out def _followup_block(items): # items: [(block_title, [missing sub_title])] block_texts = [] for title, subs in items: lines = [] for s in subs: z = f"- {s}" fk = facts_by.get((title, _norm_title(s))) if fk and (ft := _facts_lines(fk)): z += "\n" + "\n".join(" " + l for l in ft.split("\n")) lines.append(z) block_texts.append(f"BLOCK: {title}\nSUBBAUSTEINE:\n" + "\n".join(lines)) return "\n\n".join(block_texts) async def _request_more(round_n, pi, items): fp = work_dir / f"question-pattern-nach{round_n}-c{pi}.json" if _question_pattern_chunk_schema(_json_file(fp)): return # resume subs_total = sum(len(s) for _, s in items) await run_single_slot( ctx, f"Question pattern catch-up R{round_n}/{pi}", key=f"blocks-{topic}-question-pattern-nach{round_n}-c{pi}", prompt=_prompt("Question-Pattern-Research", topic=topic, blocks=_followup_block(items), out_path=fp, extra=_extra(instructions)), role="fast", capabilities="files", payload=lambda result, p=fp: _question_pattern_chunk_schema(_json_file(p)), timeout=_timeout("question_pattern", subs_total), ) for round_n in range(1, QUESTION_MAX_ROUNDS + 1): missing_subs = _missing_subs() if not missing_subs: break n_subs = sum(len(s) for _, s in missing_subs) _log(topic, f"Question pattern round {round_n}: {n_subs} sub(s) in {len(missing_subs)} block(s) without a pattern — re-request") packages = _lpt_chunks([len(s) for _, s in missing_subs], QUESTION_CHUNK_SUBS) package_items = [[missing_subs[i] for i in idxs] for idxs in packages] await _gather_progress( [_request_more(round_n, pi, items) for pi, items in enumerate(package_items)], len(package_items), _report_p(set_p, topic, "Questions check")) if is_cancelled(): return None # parse output per package + merge newly gained subs (don't overwrite existing ones). for pi, items in enumerate(package_items): title_subs = {t: subs for t, subs in items} ctitle = list(title_subs.keys()) for e in _question_pattern_chunk_schema(_json_file(work_dir / f"question-pattern-nach{round_n}-c{pi}.json")) or []: title = _match_sub(e["block"], ctitle) if title not in title_subs: continue sub = _match_sub(e["subblock"], title_subs[title]) have_set = {_norm_title(x["subblock"]) for x in outcome.get(title) or []} if _norm_title(sub) in have_set: continue outcome.setdefault(title, []).append({"subblock": sub, "question": e["question"]}) rest = _missing_subs() if rest: n = sum(len(s) for _, s in rest) _log(topic, f"Question pattern: {n} sub(s) in {len(rest)} block(s) remain empty after {QUESTION_MAX_ROUNDS} rounds: {[t for t, _ in rest][:5]}") return outcome # ── Inventory in the DB: research loop · consolidation · clarification ──────────── def _crawl_index(folder) -> dict[str, str]: """Alias (filename OR QUELLE: URL, lowercase) → canonical page key (filename).""" idx: dict[str, str] = {} if not folder or not Path(folder).is_dir(): return idx for p in sorted(Path(folder).glob("*.txt")): key = p.name idx[key.lower()] = key try: first_line = p.read_text(encoding="utf-8").splitlines()[0] except (OSError, IndexError): first_line = "" if first_line.startswith("QUELLE:"): url = first_line[len("QUELLE:"):].strip() if url: idx[url.lower()] = key idx[url.rstrip("/").lower()] = key return idx async def _set_inventory(topic: str, record: str, status: str) -> None: """Write an inventory entry ('title — description') with status to the DB.""" title = _title(record) norm = _norm_title(title) if not norm: return split_parts = [t.strip() for t in record.split(" — ")] desc = split_parts[1] if len(split_parts) >= 2 else "" await db.upsert_block(topic, norm, title, desc) await db.set_block_status(topic, norm, status) def _triage_rules(folder, pages: list[str]) -> tuple[list[str], list[str]]: """Deterministic content/noise filter (config.CRAWL_*). Substring match (lowercase) against URL + filename. Order: keep > noise > min_chars > keep. → (content, noise).""" folder = Path(folder) content, noise = [], [] for fn in pages: lines = _read(folder / fn).splitlines() url = lines[0][len("QUELLE:"):].strip() if lines and lines[0].startswith("QUELLE:") else "" body = "\n".join(lines[1:]).strip() hay = f"{url}\n{fn}".lower() if any(p in hay for p in CRAWL_KEEP_PATTERNS): content.append(fn) elif any(p in hay for p in CRAWL_NOISE_PATTERNS): noise.append(fn) elif len(body) < CRAWL_MIN_CHARS: noise.append(fn) else: content.append(fn) # default: keep — everything with content stays return content, noise def _page_snippet(folder, fn: str) -> tuple[str, str]: """(url, snippet) of a crawl page for the relevance gate. url from the QUELLE: line; snippet = body excerpt (navigation boilerplate is up front — the prompt ignores it). The URL is the primary signal (meaningful slug), the snippet only supports it.""" lines = _read(Path(folder) / fn).splitlines() url = lines[0][len("QUELLE:"):].strip() if lines and lines[0].startswith("QUELLE:") else "" body = "\n".join(lines[1:]).strip() snippet = " ".join(body.split())[:QUELLE_RELEVANZ_SNIPPET] return (url or fn), snippet async def _relevance_triage(ctx: GenContext, set_p, files: dict, folder, content: list[str], spec: str, instructions: str) -> tuple[list[str], list[str]]: """LLM topic gate after the rule filter: each content page ja/nein against the spec. Off-topic (different field) → out. Pattern like `_relevance_block`: small packages, 3 raters (`fast`), 2-of-3 consensus. CONSERVATIVE: drop only on a clear "nein" majority; dispute/gap/ race error → keep. SAFETY: if the gate would drop ≥80 % (or all), everything stays (a spec mismatch/bug must not empty the source). → (kept, out) as filenames.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] pages = sorted(content) if not pages: return content, [] items = [_page_snippet(folder, fn) for fn in pages] # index aligns with `pages` chunks = _chunk_nums(list(range(len(pages))), _n_chunks(len(pages), QUELLE_RELEVANZ_CHUNK)) n = len(chunks) def rater_paths(c): return [work_dir / f"source-relevance-c{c}-{i}.json" for i in (1, 2, 3)] def lset(idxs): return set(range(1, len(idxs) + 1)) async def _rate(c, idxs): local_set = lset(idxs) paths = rater_paths(c) existing = sum(1 for p in paths if _yesno_schema(_json_file(p), local_set)) if existing >= 2: return True enum_lines = [] for k, j in enumerate(idxs, 1): url, snip = items[j] enum_lines.append(f"{k}. {url}") if snip: enum_lines.append(f" {snip}") enum = "\n".join(enum_lines) pending = [(i, p) for i, p in enumerate(paths, 1) if not _yesno_schema(_json_file(p), local_set)] slots = [{ "key": f"blocks-{topic}-source-relevance-c{c}-{i}", "prompt": _prompt("Source-Relevance", topic=topic, spec=spec, pages=enum, out_path=p, extra=_extra(instructions)), "role": "fast", "capabilities": "files", "payload": (lambda result, p=p, ids=local_set: _yesno_schema(_json_file(p), ids)), } for i, p in pending] new = await _race(topic, f"Relevance triage package {c}", slots, 2 - existing, _timeout("relevance", len(idxs)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) return not is_cancelled() and new is not None _qidx = _step_idx(topic, "Source prep") # gate runs in the source step (no own step) set_p(f"Check relevance against spec ({n} packages)…", step=_qidx) async def _report_triage(d, t): set_p(f"Check relevance against spec {d}/{t}…", step=_qidx) await _gather_progress([_rate(c, idxs) for c, idxs in enumerate(chunks, 1)], n, _report_triage) if is_cancelled(): return content, [] # cancel → drop nothing (caller aborts) # Vote per page: only a clear "nein" majority (≥2 and more than "ja") throws it out. dropped: list[str] = [] for c, idxs in enumerate(chunks, 1): local_set = lset(idxs) rater = [d for p in rater_paths(c) if (d := _yesno_schema(_json_file(p), local_set))] for k in range(1, len(idxs) + 1): vote_list = [d[k] for d in rater if k in d] nein, ja = vote_list.count("nein"), vote_list.count("ja") if nein >= 2 and nein > ja: dropped.append(pages[idxs[k - 1]]) if dropped and len(dropped) >= max(1, int(len(pages) * 0.8)): _log(topic, f"Relevance triage: would drop {len(dropped)}/{len(pages)} — discarded (spec mismatch?), keeping all") return content, [] dropped_set = set(dropped) keepers = [fn for fn in pages if fn not in dropped_set] return keepers, dropped async def _prepare_source(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str) -> bool: """Step "Source prep": crawl (link) + PDF convert + content/noise triage. Persists the triage in the coverage table (content). → True (ok) / False (cancel/error). thema: nothing. projekt/uni: only PDFs (curated folder, no triage).""" topic, is_cancelled = ctx.topic, ctx.is_cancelled if not folder: return True # thema → no source to prepare if q["type"] != "link": await asyncio.to_thread(_convert_pdfs, folder) # projekt/uni: only PDFs, no triage return True if await db.get_step_status(topic, "Source prep") == "done": return True if not _crawl_done(topic): set_p("Loading source (crawl)…", step=_step_idx(topic, "Source prep")) n = await asyncio.to_thread(crawl, q["location"], folder, cancelled=is_cancelled) if is_cancelled(): return False if not n: _blocks_errors[topic] = "Crawl yielded no content — check link/domain" return False await asyncio.to_thread(_convert_pdfs, folder) pages = sorted(set(_crawl_index(folder).values())) if pages: set_p("Triaging pages…", step=_step_idx(topic, "Source prep")) await db.delete_coverage(topic) content, noise = _triage_rules(folder, pages) # deterministic rule filter if q.get("spec") and content: # topic gate: separates the field (rules can't) content, dropped = await _relevance_triage(ctx, set_p, files, folder, content, q["spec"], instructions) if is_cancelled(): return False if dropped: noise = sorted(set(noise) | set(dropped)) _log(topic, f"LLM relevance: {len(dropped)} pages off-topic → noise") await db.mark_content(topic, sorted(content), sorted(noise)) _log(topic, f"Triage: {len(content)} content / {len(noise)} noise of {len(pages)} (rules + LLM gate)") await db.set_step_status(topic, "Source prep", "done") return True async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str) -> bool: """Fills DB table `blocks` with candidates (+ mention counter). FIXED file batches: each crawl page is assigned to exactly one batch and read by RESEARCH_READERS agents (consensus ≥2 in the batch). All assigned pages are marked as read → 100 % coverage. Without a crawl folder (source "thema") → free web research, one round. → True/False.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled if await db.get_step_status(topic, "Research") == "done": return True work_dir = files["arbeit"] await db.delete_blocks(topic) # coverage/content belongs to the triage — do NOT delete await db.set_step_status(topic, "Research", "running") async def _ingest(reader_id: str, text: str) -> None: seen_set = set() for record in _parse_selection(text).values(): title = _title(record) norm = _norm_title(title) if not norm or norm in seen_set: continue seen_set.add(norm) # one reader = one vote per concept split_parts = [t.strip() for t in record.split(" — ")] desc = split_parts[1] if len(split_parts) >= 2 else "" source = [split_parts[2]] if len(split_parts) >= 3 and split_parts[2] else [] await db.upsert_block(topic, norm, title, desc, source, reader=reader_id) pages = await db.list_content(topic) # pages marked as content by the triage if not pages and folder: pages = sorted(set(_crawl_index(folder).values())) # fallback (projekt/uni: no triage) if not pages: # source "thema" (or no crawl): free web research, one round. set_p("Research running…", step=_step_idx(topic, "Research")) caps = "files" if folder else "full" paths = [work_dir / f"research-{i}.md" for i in range(1, RESEARCH_THEMA_AGENTS + 1)] for p in paths: p.unlink(missing_ok=True) slots = [{ "key": f"blocks-{topic}-research-{i}", "prompt": _build_research_prompt(topic, p, instructions, q["type"], folder), "role": "quick", "capabilities": caps, "payload": (lambda result, p=p, rid=f"t{i}": ((rid, t) if (t := _file_payload(p)) else None)), } for i, p in enumerate(paths, 1)] agent_texts = await _race(topic, "Research", slots, 3, _timeout("research"), provider, cancelled=is_cancelled, grace=RESEARCH_GRACE) if is_cancelled(): return False if not agent_texts: _blocks_errors[topic] = "Research failed (minimum not reached)" return False for rid, text in agent_texts: await _ingest(rid, text) await db.set_step_status(topic, "Research", "done") return True # uni/projekt: curated, often LARGE files (script). Instead of reading all at once # (lost-in-the-middle), chunk into sections and have 2 readers thoroughly read EACH — # text directly in the prompt (small context), mentions accumulate to consensus. if q["type"] in ("uni", "projekt"): eintraege: list[tuple[str, str]] = [] # (filename, section text) for fn in sorted(pages): for section_text in _text_sections(_read(folder / fn)): eintraege.append((fn, section_text)) if not eintraege: _blocks_errors[topic] = "Research: source empty" return False set_p(f"Research ({len(eintraege)} sections)…", step=_step_idx(topic, "Research")) async def _read_section(ei: int, fn: str, section_text: str) -> None: block = (f"ARBEITE AUSSCHLIESSLICH MIT DIESEM TEXTABSCHNITT (Source: {fn}). Lies ihn " f"VOLLSTÄNDIG, überspringe nichts. Notiere `{fn}` als Source jedes Bausteins. " f"Suche NICHT im Web — nur dieser Section zählt.\n\n-----\n{section_text}\n-----") paths = [work_dir / f"research-a{ei}-{i}.md" for i in range(1, RESEARCH_READERS + 1)] # reader file reuse: if all reader outputs are present and valid (resume / # re-run without research change), re-ingest instead of spawning agents again. existing = [(f"a{ei}-{i}", t) for i, p in enumerate(paths, 1) if (t := _file_payload(p))] if len(existing) == len(paths): for rid, text in existing: await _ingest(rid, text) return for p in paths: p.unlink(missing_ok=True) if is_cancelled(): return slots = [{ "key": f"blocks-{topic}-research-a{ei}-{i}", "prompt": _build_research_prompt(topic, p, instructions, q["type"], folder, section=block), "role": "quick", "capabilities": "files", "payload": (lambda result, p=p, rid=f"a{ei}-{i}": ((rid, t) if (t := _file_payload(p)) else None)), } for i, p in enumerate(paths, 1)] # quorum 2: both readers per section should pass (more eyes = more concepts + # real consensus); after timeout _race falls back to what exists. agent_texts = await _race(topic, f"Research section {ei}", slots, 2, _timeout("research", 1), provider, cancelled=is_cancelled, grace=RESEARCH_GRACE) for rid, text in (agent_texts or []): await _ingest(rid, text) await _gather_progress([_read_section(ei, fn, a) for ei, (fn, a) in enumerate(eintraege, 1)], len(eintraege), _report_p(set_p, topic, "Research")) if is_cancelled(): return False await db.mark_sources_read_done(topic, sorted(pages)) total = len(await db.list_blocks(topic)) _log(topic, f"Research (uni/projekt): {total} candidates from {len(eintraege)} sections ({len(pages)} files)") if not total: _blocks_errors[topic] = "Research failed (no blocks)" return False await db.set_step_status(topic, "Research", "done") return True # Crawl/link: many small content pages (triage in the "Source prep" step). # Fixed batches, RESEARCH_READERS readers per batch reading EXACTLY these files. batches = _chunk_nums(sorted(pages), max(1, math.ceil(len(pages) / RESEARCH_BATCH))) async def _read_batch(bi: int, batch: list[str]) -> bool: 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}") paths = [work_dir / f"research-b{bi}-{i}.md" for i in range(1, RESEARCH_READERS + 1)] for p in paths: p.unlink(missing_ok=True) if not is_cancelled(): slots = [{ "key": f"blocks-{topic}-research-b{bi}-{i}", "prompt": _build_research_prompt(topic, p, instructions, q["type"], folder, focus=fokus), "role": "quick", "capabilities": "files", "payload": (lambda result, p=p, rid=f"b{bi}-{i}": ((rid, t) if (t := _file_payload(p)) else None)), } for i, p in enumerate(paths, 1)] agent_texts = await _race(topic, f"Research batch {bi}", slots, 1, _timeout("research", len(batch)), provider, cancelled=is_cancelled, grace=RESEARCH_GRACE) for rid, text in (agent_texts or []): await _ingest(rid, text) await db.mark_sources_read_done(topic, batch) # tick off all assigned pages (even without hits) return not is_cancelled() await _gather_progress([_read_batch(bi, b) for bi, b in enumerate(batches, 1)], len(batches), _report_p(set_p, topic, "Research")) if is_cancelled(): return False total = len(await db.list_blocks(topic)) coverage = len(await db.list_coverage(topic)) _log(topic, f"Research: {total} candidates, coverage {coverage}/{len(pages)} pages ({len(batches)} batches)") if not total: _blocks_errors[topic] = "Research failed (no blocks)" return False await db.set_step_status(topic, "Research", "done") return True def _grp_schema(data, ids: set[int]): """{"groups": [[1,3],[2], …]} → partition of `ids` as a list of index groups. Tolerant: ignores foreign/duplicate numbers; forgotten candidates are added standalone (singleton group). None only on structurally broken JSON.""" if not isinstance(data, dict) or not isinstance(data.get("groups"), list): return None groups, seen_set = [], set() for g in data["groups"]: if not isinstance(g, list): return None grp = [] for x in g: try: num = int(x) except (ValueError, TypeError): continue if num in ids and num not in seen_set: seen_set.add(num) grp.append(num) if grp: groups.append(grp) groups += [[r] for r in sorted(ids - seen_set)] # forgotten candidates stay standalone return groups or None _ASPECT_MARKER = ("∈ np", "∈np", " in np", "np-schwer", "np-vollständig", "verifizierer", "zertifikat", "ndtm", "nicht-determ", "lower bound", "untere schranke", "bzgl", "als sprache") def _aspect_marker(title: str) -> int: """Number of property markers in the title (∈NP, NP-hard, verifier, lower bound …). 0 = generic main concept (the problem itself); >0 = a property of it.""" t = title.casefold() return sum(1 for m in _ASPECT_MARKER if m in t) _REFERENCE_RE = re.compile(r'^(Satz|Lemma|Korollar|Bemerkung|Definition)\s*[\d.]+\s*(\([a-z]\)|[a-z])?\s*$', re.I) def _is_reference(title: str) -> bool: """True for pure reference/placeholder titles WITHOUT meaningful content: "Satz 7.18", "Lemma 6.2", "Korollar 6.18" (number without a name) as well as marked spots "Bedingung (**)". NOT "Satz 6.24: Cook/Levin" (has a name) and NOT short technical symbols like "P⊆NP"/"Σ*" (real concepts).""" t = title.strip() if _REFERENCE_RE.match(t): return True if re.search(r'\(\*+\)', t): # marked spot "(**)" / "(*)" return True return False def _canonical(candidates: list[dict], idxs: list[int], seen_norm: set[str]) -> dict: """Representative of a cluster = the main concept (fewest property markers — the problem itself, not "… ∈ NP"); tie → most frequent norm title → most readers. Title globally unique (suffix ' (2)') so it works as a key.""" by_norm: dict[str, list[int]] = {} for k in idxs: by_norm.setdefault(_norm_title(candidates[k]["title"]), []).append(k) def weight(nb: str): ms = by_norm[nb] reader = set().union(*[set(candidates[m]["reader"]) for m in ms]) if ms else set() # reference/placeholder titles ("Satz 7.18") last — prefer a meaningful member. is_real = not _is_reference(candidates[ms[0]]["title"]) return (is_real, -_aspect_marker(nb), len(ms), len(reader)) best = max(by_norm, key=weight) k = max(by_norm[best], key=lambda m: len(candidates[m]["description"])) title = candidates[k]["title"] n = 2 while _norm_title(title) in seen_norm: title = f"{candidates[k]['title']} ({n})" n += 1 seen_norm.add(_norm_title(title)) return {"title": title, "description": candidates[k]["description"]} async def _group_blocks(ctx: GenContext, set_p, work_dir: Path, candidates: list[dict], blocks: list[list[int]], prefix: str = "consolidation", step: str = "Consolidation") -> list[list[int]]: """Per similarity block, a judge groups the titles into the real blocks (merge paraphrases, split over-merges). Singletons directly. Error/timeout → conservatively each candidate alone (avoids false over-merging). → final groups (global indices). `praefix`/`step` separate consolidation and dedup (artefacts, race key, progress).""" topic, is_cancelled = ctx.topic, ctx.is_cancelled multi = [(bi, b) for bi, b in enumerate(blocks) if len(b) > 1] outcome: list[list[int]] = [list(b) for b in blocks if len(b) == 1] # singletons directly def _line(k: int, g: int) -> str: b = candidates[g] return f"{k}. {b['title']}" + (f" — {b['description']}" if b["description"] else "") async def _grp(bi: int, block: list[int]) -> None: ids = set(range(1, len(block) + 1)) p = work_dir / f"{prefix}-block-c{bi}.json" part = _grp_schema(_json_file(p), ids) if part is None: # resume: don't recompute a valid file p.unlink(missing_ok=True) if is_cancelled(): return lines = [_line(k, block[k - 1]) for k in range(1, len(block) + 1)] status, part = await run_single_slot( ctx, f"Block grouping {bi}", key=f"blocks-{topic}-{prefix}-block-c{bi}", prompt=_prompt("Blocks-Block-Grouping", topic=topic, entries="\n".join(lines), out_path=p), role="judge", capabilities="files", payload=(lambda result, p=p, ids=ids: _grp_schema(_json_file(p), ids)), timeout=_timeout("research_mapping", len(block)), ) part = part if status == OK else None if part is None: # judge failed → individually (no over-merge) outcome.extend([idx] for idx in block) else: # local numbers → global candidate indices outcome.extend([block[k - 1] for k in g] for g in part) await _gather_progress([_grp(bi, b) for bi, b in multi], len(multi), _report_p(set_p, topic, step)) return outcome async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates: list[dict]) -> bool: """Two-stage: embeddings → coarse capped blocks (high recall) → one judge per multi-block, grouping the titles into the real blocks → reader union (≥2 = consensus).""" topic, is_cancelled = ctx.topic, ctx.is_cancelled work_dir = files["arbeit"] texts = [f"{b['title']} — {b['description']}" if b["description"] else b["title"] for b in candidates] sims = await asyncio.to_thread(embedding.embed_sims, texts) if sims is None: # model not available after all → fallback return await _consolidate_llm(ctx, set_p, files, candidates) # Level 1: coarse similarity blocks (capped, no giant component). blocks = await asyncio.to_thread(embedding.capped_blocks, sims, None, None) # Level 2: one judge groups EACH multi-block into the real blocks. groups = await _group_blocks(ctx, set_p, work_dir, candidates, blocks) if is_cancelled(): return False def _min_cos(idxs): # internal coherence as a check (chains would be ~0.3) if len(idxs) < 2: return 1.0 return round(min(float(sims[i][j]) for n, i in enumerate(idxs) for j in idxs[n + 1:]), 3) # Consensus = ≥2 distinct readers per cluster. Legacy DBs without reader tracking (research ran # before the migration, no re-ingest) have empty reader sets → fall back to a title heuristic # (otherwise EVERYTHING would land in the rest). hat_reader = any(b["reader"] for b in candidates) consensus, rest, debug, seen_norm = [], [], [], set() for idxs in groups: reader = set().union(*[set(candidates[k]["reader"]) for k in idxs]) if idxs else set() if hat_reader: score = len(reader) else: # without reader data: max(mentions, number of distinct title variants in the cluster) score = max(max(candidates[k]["mentions"] for k in idxs), len({candidates[k]["title_norm"] for k in idxs})) rep = _canonical(candidates, idxs, seen_norm) record = f"{rep['title']} — {rep['description']}" if rep["description"] else rep["title"] (consensus if score >= 2 else rest).append(record) debug.append({"title": rep["title"], "reader": sorted(reader), "score": score, "consensus": score >= 2, "min_cos": _min_cos(idxs), "mitglieder": [candidates[k]["title"] for k in idxs]}) atomic_write_json(work_dir / "consolidation-cluster.json", debug, indent=1) multi_blocks = sum(1 for b in blocks if len(b) > 1) _log(topic, f"Consolidation (embedding): {len(blocks)} blocks ({multi_blocks} grouped via LLM) " f"→ {len(groups)} clusters from {len(candidates)} candidates " f"→ {len(consensus)} consensus / {len(rest)} rest") await db.delete_blocks(topic) for t in consensus: await _set_inventory(topic, t, "consensus") for t in rest: await _set_inventory(topic, t, "rest") await db.set_step_status(topic, "Consolidation", "done") return True async def _consolidate(ctx: GenContext, set_p, files: dict) -> bool: """Merges raw candidates into consensus (≥2 readers)/rest. Deterministic via embedding clustering; if the model is missing → fall back to the LLM panel (`_consolidate_llm`). Status in DB.""" topic = ctx.topic if await db.get_step_status(topic, "Consolidation") == "done": return True set_p("Consolidating research…", step=_step_idx(topic, "Consolidation")) candidates = await db.list_blocks(topic) if not candidates: _blocks_errors[topic] = "Consolidation: no candidates" return False if EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available): return await _consolidate_embedding(ctx, set_p, files, candidates) return await _consolidate_llm(ctx, set_p, files, candidates) async def _consolidate_llm(ctx: GenContext, set_p, files: dict, candidates: list[dict]) -> bool: """Fallback (only without an embedding model): a panel (KONSOLIDIERUNG_PANEL judges) merges candidates semantically; a reconcile judge combines the panel outputs into the final consensus (≥2)/rest (1×) list. Panel instead of a single judge: a single judge is bias-prone and unstable.""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] chunks = _chunk_nums(candidates, max(1, math.ceil(len(candidates) / CONSOLIDATION_CHUNK))) async def _map_panel(c: int, eintraege: str, amount: int): """3 mapping judges over `eintraege` → reconcile judge → (consensus, rest). None on cancel/error.""" paths = [work_dir / f"consolidation-c{c}-j{j}.json" for j in range(1, CONSOLIDATION_PANEL + 1)] pending = [(j, p) for j, p in enumerate(paths, 1) if _mapping_schema(_json_file(p)) is None] for _, p in pending: p.unlink(missing_ok=True) if pending: slots = [{ "key": f"blocks-{topic}-consolidation-c{c}-j{j}", "prompt": _prompt("Blocks-Research-Mapping", topic=topic, n=RESEARCH_READERS, entries=eintraege, out_path=p), "role": "judge", "capabilities": "files", "payload": (lambda result, p=p: _mapping_schema(_json_file(p))), } for j, p in pending] existing = CONSOLIDATION_PANEL - len(pending) await _race(topic, f"Consolidation {c}", slots, max(1, 2 - existing), _timeout("research_mapping", amount), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) if is_cancelled(): return None outs = [m for p in paths if (m := _mapping_schema(_json_file(p)))] if not outs: return None # union of panel titles; per title count how many judges list it as consensus. kvotes: dict[str, int] = {} form: dict[str, str] = {} # norm → display title (first occurrence) order: list[str] = [] for kk, rr in outs: for t in kk + rr: nt = _norm_title(_title(t)) if not nt: continue if nt not in form: form[nt] = t order.append(nt) kvotes.setdefault(nt, 0) for t in kk: nt = _norm_title(_title(t)) if nt: kvotes[nt] = kvotes.get(nt, 0) + 1 # Reconcile: one merge judge over the union, annotated with judge votes ("k× genannt"). rp = work_dir / f"consolidation-c{c}-reconcile.json" recon = _mapping_schema(_json_file(rp)) if recon is None: rp.unlink(missing_ok=True) entries_r = "\n".join(f"{i}. {form[nt]} ({max(1, kvotes[nt])}× genannt)" for i, nt in enumerate(order, 1)) status, recon = await run_single_slot( ctx, f"Consolidation Reconcile {c}", key=f"blocks-{topic}-consolidation-c{c}-reconcile", prompt=_prompt("Blocks-Research-Mapping", topic=topic, n=CONSOLIDATION_PANEL, entries=entries_r, out_path=rp), role="judge", capabilities="files", payload=lambda result, p=rp: _mapping_schema(_json_file(p)), timeout=_timeout("research_mapping", len(order)), ) if status == CANCELLED: return None recon = recon if status != FAILED else None if recon: return recon # Fallback (reconcile failed): code majority — consensus if a majority of judges say consensus. consensus = [form[nt] for nt in order if kvotes[nt] * 2 >= len(outs) and kvotes[nt] > 0] kset = {_norm_title(_title(t)) for t in consensus} return consensus, [form[nt] for nt in order if nt not in kset] consensus, rest = [], [] for c, chunk in enumerate(chunks, 1): eintraege = "\n".join( f"{i}. {b['title']} — {b['description']} ({b['mentions']}× genannt)" for i, b in enumerate(chunk, 1) ) res = await _map_panel(c, eintraege, len(chunk)) if res is None: if is_cancelled(): return False _blocks_errors[topic] = "Research mapping failed" return False k, r = res consensus += k rest += r # With multiple chunks: a global merge pass over the combined consensus entries, # so duplicates across chunk boundaries (DAL×4, PHPUnit×5 …) merge. if len(chunks) > 1 and consensus: fp = work_dir / "consolidation-merge.json" fp.unlink(missing_ok=True) eintraege = "\n".join(f"{i}. {t} (2× genannt)" for i, t in enumerate(consensus, 1)) status, mapping = await run_single_slot( ctx, "Consolidation Merge", key=f"blocks-{topic}-consolidation-merge", prompt=_prompt("Blocks-Research-Mapping", topic=topic, n=RESEARCH_READERS, entries=eintraege, out_path=fp), role="judge", capabilities="files", payload=lambda result, p=fp: _mapping_schema(_json_file(p)), timeout=_timeout("research_mapping", len(consensus)), ) if status == CANCELLED: return False if status != FAILED and mapping: consensus, r2 = mapping rest += r2 # entries downgraded by the merge into the rest # Judge output is authoritative → re-set the inventory in the DB. await db.delete_blocks(topic) for t in consensus: await _set_inventory(topic, t, "consensus") for t in rest: await _set_inventory(topic, t, "rest") await db.set_step_status(topic, "Consolidation", "done") return True async def _clarify_inventory(ctx: GenContext, set_p, files: dict) -> bool: """A panel (KONSOLIDIERUNG_PANEL judges) decides on the rest (1×-mentioned): majority `aufnehmen` → consensus, otherwise discarded. Panel instead of a single judge — the rest cut is the sharpest intervention; a single judge is too unstable here. Conservative tie → keep (never lose a concept).""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled if await db.get_step_status(topic, "Clarification") == "done": return True set_p("Clarification running…", step=_step_idx(topic, "Clarification")) rest_rows = await db.list_blocks(topic, status="rest") # Continuous gate (EDC "Define"): also check consensus blocks with reference/placeholder titles # ("Satz 7.18", "Korollar 6.18", "Bedingung (**)") — otherwise they bypass every exam. suspicious = [b for b in await db.list_blocks(topic, status="consensus") if _is_reference(b["title"])] check_rows = rest_rows + suspicious if check_rows: work_dir = files["arbeit"] paths = [work_dir / f"clarification-j{j}.json" for j in range(1, CONSOLIDATION_PANEL + 1)] # final=False: a judge with an accidentally non-empty `rest` must not fail entirely # (otherwise the panel collapses to 1 judge). Its `aufnehmen` counts; rest entries count as # not-accepted. The "rest empty" requirement still stands in the prompt. pending = [(j, p) for j, p in enumerate(paths, 1) if _runde_schema(_json_file(p)) is None] for _, p in pending: p.unlink(missing_ok=True) if pending: slots = [{ "key": f"blocks-{topic}-clarification-j{j}", "prompt": _prompt( "Blocks-Klaerung", topic=topic, rest="\n".join(f"- {b['title']} — {b['description']}" if b['description'] else f"- {b['title']}" for b in check_rows), final="\n- Entscheide JEDEN Eintrag. `rest` MUSS leer sein.", out_path=p, ), "role": "judge", "capabilities": "files", "payload": (lambda result, p=p: _runde_schema(_json_file(p))), } for j, p in pending] existing = CONSOLIDATION_PANEL - len(pending) await _race(topic, "Clarification", slots, max(1, 2 - existing), _timeout("selection_mapping", len(check_rows)), provider, cancelled=is_cancelled, grace=CONSENSUS_GRACE) if is_cancelled(): return False outs = [r for p in paths if (r := _runde_schema(_json_file(p)))] if not outs: _blocks_errors[topic] = "Clarification failed" return False # Majority per rest entry (by norm title). Tie → keep (votes*2 >= n). 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 # Rename suggestions (additive from the raw JSON — _runde_schema doesn't know the field): # kept reference/placeholder titles → meaningful name from the content. Old title norm # stays stable (doesn't break the votes match); per old title the most frequent suggestion. renames: dict[str, dict[str, int]] = {} for p in paths: d = _json_file(p) rename_raw = d.get("rename") if isinstance(d, dict) else None if isinstance(rename_raw, dict): for old, new in rename_raw.items(): new = str(new).strip() if new: renames.setdefault(_norm_title(str(old)), {}).setdefault(new, 0) renames[_norm_title(str(old))][new] += 1 seen_norm = {b["title_norm"] for b in await db.list_blocks(topic, status="consensus")} for b in check_rows: accept = votes.get(b["title_norm"], 0) * 2 >= len(outs) if not accept: await db.set_block_status(topic, b["title_norm"], "discarded") continue new_title = None if _is_reference(b["title"]) and (suggestions := renames.get(b["title_norm"])): cands = max(suggestions, key=lambda k: (suggestions[k], len(k))) if not _is_reference(cands): new_title = cands if new_title: nn, t, n = _norm_title(new_title), new_title, 2 while nn in seen_norm: t, nn, n = f"{new_title} ({n})", _norm_title(f"{new_title} ({n})"), n + 1 seen_norm.add(nn) await db.set_block_status(topic, b["title_norm"], "consensus", title=t, neu_norm=nn) else: await db.set_block_status(topic, b["title_norm"], "consensus") await db.set_step_status(topic, "Clarification", "done") return True def _pairs_schema(data) -> dict[int, bool] | None: """{"pairs": {"1": "ja", "2": "nein", …}} → {pair_nr: True/False} · otherwise None.""" if not isinstance(data, dict) or not isinstance(data.get("pairs"), dict): return None out: dict[int, bool] = {} for k, v in data["pairs"].items(): try: nr = int(k) except (ValueError, TypeError): continue out[nr] = str(v).strip().casefold() in ("ja", "yes", "true", "1") return out or None def _cliques(n: int, edge_list: list[tuple[int, int]]) -> list[list[int]]: """Complete-link: greedy maximal cliques over the confirmed duplicate edges. A group forms only if ALL its nodes are pairwise connected → no chaining (A=B + B=C forms NO group {A,B,C} as long as A=C is missing). Only cliques ≥2 are returned.""" adj: dict[int, set[int]] = {i: set() for i in range(n)} for a, b in edge_list: adj[a].add(b) adj[b].add(a) used: set[int] = set() groups: list[list[int]] = [] for v in sorted(range(n), key=lambda x: -len(adj[x])): if v in used or not adj[v]: continue clique = {v} for u in sorted(adj[v], key=lambda x: -len(adj[x])): if u not in used and clique <= adj[u] | {u}: # u connected to ALL previous ones clique.add(u) if len(clique) >= 2: groups.append(sorted(clique)) used |= clique return groups async def _dedup_inventory(ctx: GenContext, set_p, files: dict) -> bool: """Final dedup pass over the finished consensus list: pairwise verification (entity resolution). Embedding yields candidate PAIRS (cosine ≥ DEDUP_PAAR_FLOOR), a judge confirms EACH pair individually (ja = the same duplicate). ONLY confirmed pairs become merge edges (union-find) — no chaining, no aspect over-merging like the block mixer. Per group ONE representative (main concept) stays, the rest is discarded.""" topic, is_cancelled = ctx.topic, ctx.is_cancelled if await db.get_step_status(topic, "Dedup") == "done": return True if not (EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available)): await db.set_step_status(topic, "Dedup", "done") # without a model: silently skip return True set_p("Dedup…", step=_step_idx(topic, "Dedup")) work_dir = files["arbeit"] consensus = await db.list_blocks(topic, status="consensus") if len(consensus) >= 2: import numpy as np texts = [f"{b['title']} — {b['description']}" if b["description"] else b["title"] for b in consensus] sims = await asyncio.to_thread(embedding.embed_sims, texts) if sims is not None: n = len(consensus) iu = np.triu_indices(n, k=1) cands = [(int(iu[0][m]), int(iu[1][m])) for m in np.where(sims[iu] >= DEDUP_PAIR_FLOOR)[0]] _log(topic, f"Dedup: {len(cands)} candidate pairs (cosine ≥ {DEDUP_PAIR_FLOOR}) → pairwise filter") packages = [cands[i:i + DEDUP_PAIRS_CHUNK] for i in range(0, len(cands), DEDUP_PAIRS_CHUNK)] def pair_path(pi): return work_dir / f"dedup-paar-c{pi}.json" async def _filt(pi, paare): fp = pair_path(pi) if _pairs_schema(_json_file(fp)): return # resume lines = "\n\n".join( f"{j + 1}.\nA: {consensus[a]['title']} — {consensus[a]['description']}" f"\nB: {consensus[b]['title']} — {consensus[b]['description']}" for j, (a, b) in enumerate(paare)) await run_single_slot( ctx, f"Dedup pairs {pi}", key=f"blocks-{topic}-dedup-paar-c{pi}", prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=fp), role="judge", capabilities="files", payload=lambda result, p=fp: _pairs_schema(_json_file(p)), timeout=_timeout("selection_mapping", len(paare)), ) await _gather_progress([_filt(pi, p) for pi, p in enumerate(packages)], len(packages), _report_p(set_p, topic, "Dedup")) if is_cancelled(): return False # Collect confirmed "ja" edges, then COMPLETE-LINK (greedy cliques) instead of single-link # union-find — prevents chaining (A=B + B=C does NOT merge A,C without a direct A=C). edge_list, ja = [], 0 for pi, paare in enumerate(packages): verdict = _pairs_schema(_json_file(pair_path(pi))) or {} for j, (a, b) in enumerate(paare): if verdict.get(j + 1): edge_list.append((a, b)) ja += 1 groups = _cliques(n, edge_list) removed = 0 for idxs in groups: # representative = main concept (fewest property markers), then shortest title. rep = min(idxs, key=lambda k: (_aspect_marker(consensus[k]["title"]), len(consensus[k]["title"]), k)) for k in idxs: if k != rep: await db.set_block_status(topic, consensus[k]["title_norm"], "discarded") removed += 1 from collections import Counter atomic_write_json(work_dir / "dedup-runde-1.json", {"vorher": n, "entfernt": removed, "paare_geprueft": len(cands), "paare_ja": ja, "clique_groessen": dict(sorted(Counter(len(g) for g in groups).items())), "groups": [[consensus[k]["title"] for k in g] for g in groups]}, indent=1) _log(topic, f"Dedup (pairwise): {n} → {n - removed} (−{removed}); {ja}/{len(cands)} pairs confirmed") await db.set_step_status(topic, "Dedup", "done") return True def _filter_schema(data) -> dict[int, int] | None: """{"fragments": {"3": 7, "12": 8}} → {block_nr: parent_nr} · None on invalid structure. Empty dict = valid (nothing to degrade). Parent ≠ itself.""" if not isinstance(data, dict) or not isinstance(data.get("fragments"), dict): return None out: dict[int, int] = {} for k, v in data["fragments"].items(): try: nr, parent = int(k), int(v) except (ValueError, TypeError): continue if nr != parent: out[nr] = parent return out # Pure notation/symbols without a standalone concept — kept narrow (FP~0, checked against aak; # "KNF"/"MST"/"NP" do NOT match). These are discarded autonomously (need no parent). _FILTER_NOTATION = re.compile(r'^\s*\|.{1,6}\|\s*$|^Güte\s+\d+\s*$') # Property/runtime suspicion — marks lines for the judge's verdict (NO auto-drop, FP too high: # "NP-Schwere", reductions with "∈NP" are real blocks). Complements _aspekt_marker. _FILTER_PREDICATE = re.compile( r'ist NP-(vollständig|schwer)|NP-(Vollständigkeit|Schwere) von|ETH (Konsequenz|Lower Bound)' r'|Approximationsschema nach|Laufzeit O\(|∈ ?NP', re.I) def _filter_suspect(b: dict) -> bool: """Heuristic flag: could be a property/detail of another block.""" return _aspect_marker(b["title"]) > 0 or bool(_FILTER_PREDICATE.search(f"{b['title']} {b['description'] or ''}")) async def _filter_inventory(ctx: GenContext, set_p, files: dict) -> bool: """Degrade pass (granularity): separates real blocks from fragments (properties, proof gadgets, notation, runtime details). Each judge sees the FULL block list (self-containment is relational) and marks fragments WITH a parent block from the list. Fragment + parent-in-list → discarded (content comes back as a subblock of the parent). No parent or in doubt → keep (no concept loss).""" topic, is_cancelled = ctx.topic, ctx.is_cancelled if await db.get_step_status(topic, "Blocks-Filter") == "done": return True set_p("Blocks-Filter…", step=_step_idx(topic, "Blocks-Filter")) work_dir = files["arbeit"] consensus_all = await db.list_blocks(topic, status="consensus") # Safety net: discard pure notation autonomously (FP~0, no parent needed). The judge # reliably overlooks such symbols (recall problem), hence deterministically beforehand. consensus, notation_dropped = [], [] for b in consensus_all: if _FILTER_NOTATION.search(b["title"]): await db.set_block_status(topic, b["title_norm"], "discarded") notation_dropped.append(b["title"]) else: consensus.append(b) if notation_dropped: _log(topic, f"Blocks-Filter: {len(notation_dropped)} pure notation discarded: {notation_dropped[:6]}") if len(consensus) < 2: await db.set_step_status(topic, "Blocks-Filter", "done") return True n = len(consensus) # ⚠ marks suspicious lines (property/runtime) — the judge MUST check them per entry. def _line(i, b): mark = "⚠ " if _filter_suspect(b) else "" return f"{i}. {mark}{b['title']} — {b['description']}" if b["description"] else f"{i}. {mark}{b['title']}" full_list = "\n".join(_line(i, b) for i, b in enumerate(consensus, 1)) chunks = [list(range(i, min(i + FILTER_CHUNK, n + 1))) for i in range(1, n + 1, FILTER_CHUNK)] def filt_path(ci): return work_dir / f"inventar-filter-c{ci}.json" async def _assess(ci, numbers): fp = filt_path(ci) if _filter_schema(_json_file(fp)) is not None: return # resume await run_single_slot( ctx, f"Blocks-Filter {ci}", key=f"blocks-{topic}-inventar-filter-c{ci}", prompt=_prompt("Blocks-Filter", topic=topic, list=full_list, from_n=numbers[0], to_n=numbers[-1], out_path=fp), role="judge", capabilities="files", payload=lambda result, p=fp: _filter_schema(_json_file(p)), timeout=_timeout("selection_mapping", len(numbers)), ) await _gather_progress([_assess(ci, nm) for ci, nm in enumerate(chunks)], len(chunks), _report_p(set_p, topic, "Blocks-Filter")) if is_cancelled(): return False fragments: dict[int, int] = {} for ci, numbers in enumerate(chunks): verdict = _filter_schema(_json_file(filt_path(ci))) or {} nset = set(numbers) for nr, parent in verdict.items(): if 1 <= parent <= n and nr in nset: fragments[nr] = parent # Chain protection: a block that is itself the parent of a fragment stays (its child needs the anchor). parent_set = set(fragments.values()) removed, debug = 0, [] for nr, parent in fragments.items(): if nr in parent_set: continue b = consensus[nr - 1] await db.set_block_status(topic, b["title_norm"], "discarded") removed += 1 debug.append({"fragment": b["title"], "eltern": consensus[parent - 1]["title"]}) atomic_write_json(work_dir / "inventar-filter.json", {"vorher": n, "degradiert": removed, "fragments": debug}, indent=1) _log(topic, f"Blocks-Filter: {n} → {n - removed} (−{removed} fragments → subblocks)") await db.set_step_status(topic, "Blocks-Filter", "done") return True # --- Outline (blocks artifact: chapter structure, only read by the guide) --- def _outline_complete(files: dict) -> bool: """Is the outline present (chapter list exists)?""" d = _json_file(files["outline"]) return isinstance(d, dict) and isinstance(d.get("chapters"), list) and bool(d.get("chapters")) def _outline_schema(data, valid: set[int]): """{"chapters":[{title,numbers}]} → cleaned (valid numbers, each exactly once) · None at <80 % coverage (agent/judge omitted too much).""" if not isinstance(data, dict) or not isinstance(data.get("chapters"), list): return None out, seen = [], set() for ch in data["chapters"]: if not isinstance(ch, dict): continue title = str(ch.get("title", "")).strip() or "Chapter" nums = [] for n in (ch.get("numbers") or []): try: n = int(n) except (ValueError, TypeError): continue if n in valid and n not in seen: seen.add(n) nums.append(n) if nums: out.append({"title": title, "numbers": nums}) if not out or len(seen) < 0.8 * len(valid): return None return {"chapters": out} def _prereq_schema(data, valid: set[int]) -> dict[int, list[int]]: """{"prereqs": {"3": [1, 7]}} → {num: [prereq nums]} · only numbers from `valid`, no self-edge. Invalid/empty → {} (best-effort: then original order).""" if not isinstance(data, dict) or not isinstance(data.get("prereqs"), dict): return {} out: dict[int, list[int]] = {} for k, v in data["prereqs"].items(): try: num = int(k) except (ValueError, TypeError): continue if num not in valid or not isinstance(v, list): continue pres = [] for p in v: try: p = int(p) except (ValueError, TypeError): continue if p in valid and p != num and p not in pres: pres.append(p) if pres: out[num] = pres return out def _topo_order(nums: list[int], edges: dict[int, list[int]]) -> list[int]: """Kahn topo sort: prerequisites first. `edges[num]` = numbers that must come BEFORE num. Stable tie-break (original order of `nums`); cycles are broken (never deadlock).""" pos = {n: i for i, n in enumerate(nums)} # remaining in-degree over valid nodes only; self/foreign edges ignored. pre = {n: [p for p in edges.get(n, []) if p in pos and p != n] for n in nums} done: list[int] = [] finished: set[int] = set() rest = list(nums) while rest: ready_nodes = [n for n in rest if all(p in finished for p in pre[n])] if not ready_nodes: # cycle → force the earliest remaining node in original order ready_nodes = [min(rest, key=lambda n: pos[n])] nxt = min(ready_nodes, key=lambda n: pos[n]) # stable: smallest original position first done.append(nxt) finished.add(nxt) rest.remove(nxt) return done async def _learning_order(ctx: GenContext, set_p, files: dict, entries: dict, valid: set[int], instructions: str) -> dict: """Put entries (num→title) into learning order: the LLM extracts prereq edges from the extracted `prerequisites`, code solves via topo sort. Best-effort → otherwise entries unchanged.""" if len(entries) < 3: return entries topic = ctx.topic facts_map = _json_file(files["facts"]) facts_map = facts_map if isinstance(facts_map, dict) else {} def _hint(title): fm = facts_map.get(title) or {} vs = [v for fk in fm.values() if isinstance(fk, dict) and (v := str(fk.get("prerequisites", "")).strip())] return " · ".join(dict.fromkeys(vs)) pp = files["arbeit"] / "outline-prereqs.json" def _payload(result, p=pp): d = _json_file(p) return d if isinstance(d, dict) and "prereqs" in d else None existing = _json_file(pp) if not (isinstance(existing, dict) and "prereqs" in existing): lines = [f"{n}. {t}" + (f"\n braucht vorher: {h}" if (h := _hint(t)) else "") for n, t in entries.items()] set_p("Outline — learning order…", step=_step_idx(topic, "Outline")) await run_single_slot( ctx, "Outline-Prerequisites", key=f"blocks-{topic}-outline-prereqs", prompt=_prompt("Outline-Prerequisites", topic=topic, blocks="\n".join(lines), out_path=pp, extra=_extra(instructions)), role="guide", capabilities="files", payload=_payload, timeout=_timeout("plan", len(entries))) edges = _prereq_schema(_json_file(pp), valid) if not edges: return entries # no/invalid edges → original order (no regression) ordered = _topo_order(list(entries), edges) return {n: entries[n] for n in ordered} async def _outline_block(ctx: GenContext, set_p, files: dict, entries: dict, instructions: str) -> dict: """Format-agnostic outline over ALL blocks — 3 proposals → judge merges. Never aborts: 0 valid → one chapter with everything; missing blocks land in "Other". → {"chapters":[{title,numbers}]} (also in files["outline"]).""" topic, is_cancelled = ctx.topic, ctx.is_cancelled valid = set(entries) step = _step_idx(topic, "Outline") # Establish learning order (LLM-modulo): the LLM extracts prereq edges from the extracted # `prerequisites`, code solves via topo sort. Best-effort → otherwise original order. entries = await _learning_order(ctx, set_p, files, entries, valid, instructions) liste = "\n".join(f"{n}. {t}" for n, t in entries.items()) set_p("Outline — proposals…", step=step) async def _proposal(i, path): if _outline_schema(_json_file(path), valid): return True await run_single_slot( ctx, f"Outline {i}", key=f"blocks-{topic}-outline-{i}", prompt=_prompt("Guide-Outline", topic=topic, blocks=liste, out_path=path, extra=_extra(instructions)), role="guide", capabilities="files", payload=lambda result, p=path: _outline_schema(_json_file(p), valid), timeout=_timeout("plan", len(entries))) return _outline_schema(_json_file(path), valid) is not None slots = files["outline_slots"] await _gather_progress([_proposal(i, p) for i, p in enumerate(slots, 1)], len(slots), _report_p(set_p, topic, "Outline")) if is_cancelled(): return {} proposals = [v for p in slots if (v := _outline_schema(_json_file(p), valid))] if not proposals: plan = {"chapters": [{"title": "Contents", "numbers": list(entries)}]} elif len(proposals) == 1: plan = proposals[0] else: set_p("Outline merging…", step=step) block_texts = "\n\n".join( f"### Vorschlag {i}\n" + "\n".join( f"KAPITEL: {ch['title']}\n Nummern: {', '.join(str(n) for n in ch['numbers'])}" for ch in v["chapters"]) for i, v in enumerate(proposals, 1)) await run_single_slot( ctx, "Outline-Judge", key=f"blocks-{topic}-outline-judge", prompt=_prompt("Guide-Outline-Judge", topic=topic, format_name="den Guide", purpose="alle Blocks in einem roten Faden", n=len(proposals), blocks=liste, outlines=block_texts, out_path=files["outline"], extra=_extra(instructions)), role="judge", capabilities="files", payload=lambda result: _outline_schema(_json_file(files["outline"]), valid), timeout=_timeout("plan_judge", len(entries))) plan = _outline_schema(_json_file(files["outline"]), valid) or proposals[0] # Completeness: every block appears — missing in "Other" (against omitting agents/judge). included = {n for ch in plan["chapters"] for n in ch["numbers"]} missing = [n for n in entries if n not in included] if missing: plan["chapters"].append({"title": "Other", "numbers": missing}) atomic_write_json(files["outline"], plan, indent=1) return plan # --- Learning artefacts (flashcards/examples from the facts) --- def _cards_schema(data): """{"cards":[{block,subblock,question,answer}]} → list (also empty) · None if broken.""" if not isinstance(data, dict) or not isinstance(data.get("cards"), list): return None out = [] for e in data["cards"]: if isinstance(e, dict) and (f := str(e.get("question", "")).strip()) and (a := str(e.get("answer", "")).strip()): out.append({"block": str(e.get("block", "")).strip(), "subblock": str(e.get("subblock", "")).strip(), "question": f, "answer": a}) return out def _example_schema(data): """{"examples":[{block,subblock,problem,steps,result}]} → list (also empty) · None if broken.""" if not isinstance(data, dict) or not isinstance(data.get("examples"), list): return None out = [] for e in data["examples"]: if not isinstance(e, dict): continue problem = str(e.get("problem", "")).strip() steps = [s for x in (e.get("steps") or []) if (s := str(x).strip())] if problem and steps: out.append({"block": str(e.get("block", "")).strip(), "subblock": str(e.get("subblock", "")).strip(), "problem": problem, "steps": steps, "result": str(e.get("result", "")).strip()}) return out def _example_check_schema(data): """Worked-example check → {"ok": true} → set() (all correct); {"problems":[{"index":N}]} → {N, …} (1-based flagged indices); None if broken.""" if not isinstance(data, dict): return None if data.get("ok") is True: return set() pr = data.get("problems") if not isinstance(pr, list): return None out: set[int] = set() for p in pr: if isinstance(p, dict): try: out.add(int(p.get("index"))) except (ValueError, TypeError): continue return out _ARTEFACT_SCHEMA = {"flashcard": _cards_schema, "example": _example_schema} _ARTEFACT_PROMPT = {"flashcard": "Artifact-Flashcard", "example": "Artifact-Example"} _ARTEFACT_STEP = {"flashcard": "Flashcards", "example": "Examples"} def _artefacts_complete(files: dict) -> bool: """Artifact map present (all types generated)? Values may be empty (content-aware).""" d = _json_file(files["artefakte"]) return isinstance(d, dict) and all(t in d for t in ARTEFACT_TYPES) async def _artefacts_block(ctx: GenContext, set_p, files: dict, sidecar: dict, instructions: str) -> dict | None: """Generate learning artefacts per type from the stored facts — one generation pass per type over chunks. Worked examples are verified against the facts (wrong ones discarded); flashcards are low-risk and stay unchecked. → {type: [entries]} (also in files).""" topic, provider, is_cancelled = ctx.topic, ctx.provider, ctx.is_cancelled work_dir = files["arbeit"] caps = "files" # Blocks with subs + facts lines as input block (extract-once from the facts). blocks = [] for btitle, subs in sidecar.items(): if not isinstance(subs, list): continue lines = [] for s in subs: if not isinstance(s, dict) or not (st := str(s.get("title", "")).strip()): continue fk = s.get("facts") if isinstance(s.get("facts"), dict) else {} line = f"- {st}" if fk and (fk_text := _facts_lines(fk)): line += "\n" + "\n".join(" " + l for l in fk_text.split("\n")) lines.append(line) if lines: blocks.append((btitle, lines)) if not blocks: empty_map = {t: [] for t in ARTEFACT_TYPES} atomic_write_json(files["artefakte"], empty_map, indent=1) return empty_map chunks = _lpt_chunks([len(z) for _, z in blocks], FACTS_CHUNK_SUBS) def block_text(idxs): return "\n\n".join(f"BLOCK: {blocks[i][0]}\nSUBBAUSTEINE:\n" + "\n".join(blocks[i][1]) for i in idxs) # Check worked examples against the facts (panel majority) — discard wrong ones. CoT steps are # error-prone; a wrong example imprints a faulty schema → no example > a wrong one. async def _check_examples(ci, idxs, items): if is_cancelled() or not items: return items def cpath(j): return work_dir / f"artifact-example-check-c{ci}-j{j}.json" examples_txt = "\n\n".join( f"{k}. PROBLEM: {e['problem']}\n SCHRITTE: " + " | ".join(e.get("steps", [])) + (f"\n ERGEBNIS: {e['result']}" if e.get("result") else "") for k, e in enumerate(items, 1)) pending = [j for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if _example_check_schema(_json_file(cpath(j))) is None] if pending: await asyncio.gather(*[ run_agent(f"blocks-{topic}-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=caps) for j in pending], return_exceptions=True) 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) votes: dict[int, int] = {} for s in outs: for idx in s: votes[idx] = votes.get(idx, 0) + 1 threshold = len(outs) / 2 dropped = {idx for idx, v in votes.items() if v > threshold} # majority (≥2 of 3) flagged → out if dropped: _log(topic, f"Worked-example check chunk {ci}: {len(dropped)}/{len(items)} discarded") return [e for k, e in enumerate(items, 1) if k not in dropped] outcome: dict[str, list] = {} for type in ARTEFACT_TYPES: schema = _ARTEFACT_SCHEMA[type] def apath(ci, t=type): return work_dir / f"artifact-{t}-c{ci}.json" async def _gen(ci, idxs, t=type, schema=schema): p = apath(ci, t) if schema(_json_file(p)) is not None: return True await run_single_slot( ctx, f"{_ARTEFACT_STEP[t]} {ci}", key=f"blocks-{topic}-artifact-{t}-c{ci}", prompt=_prompt(_ARTEFACT_PROMPT[t], topic=topic, blocks=block_text(idxs), out_path=p, extra=_extra(instructions)), role="guide", capabilities="files", payload=lambda result, p=p, schema=schema: schema(_json_file(p)), timeout=_timeout("content", sum(len(blocks[i][1]) for i in idxs))) return schema(_json_file(p)) is not None await _gather_progress([_gen(ci, idxs) for ci, idxs in enumerate(chunks)], len(chunks), _report_p(set_p, topic, _ARTEFACT_STEP[type])) if is_cancelled(): return None eintraege: list = [] for ci in range(len(chunks)): chunk_items = schema(_json_file(apath(ci))) or [] if type == "example" and chunk_items: chunk_items = await _check_examples(ci, chunks[ci], chunk_items) eintraege += chunk_items outcome[type] = eintraege atomic_write_json(files["artefakte"], outcome, indent=1) return outcome async def _mirror_artefacts_db(topic: str, sidecar: dict, artefacts: dict) -> None: """Mirror artefacts into the DB. Flashcard/example per sub (sub_norm).""" await db.delete_sub_artefakte(topic) btitle_list = list(sidecar.keys()) for type in ARTEFACT_TYPES: for e in artefacts.get(type, []): bt = _match_sub(e.get("block", ""), btitle_list) bnorm, sn = _norm_title(bt), _norm_title(e.get("subblock", "")) if not bnorm or not sn: continue 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, type, data, bt, e.get("subblock", "")) async def _mirror_sidecar_db(topic: str, sidecar: dict) -> None: """Mirror the sidecar {block title: [{title, level, relevance}]} into the DB table subblocks.""" for btitle, subs in sidecar.items(): bnorm = _norm_title(btitle) if not bnorm or not isinstance(subs, list): continue for s in subs: if not isinstance(s, dict): continue st = str(s.get("title", "")).strip() sn = _norm_title(st) if not sn: continue facts = json.dumps(s["facts"], ensure_ascii=False) if isinstance(s.get("facts"), dict) else None await db.put_subblock(topic, bnorm, sn, btitle, st, level=s.get("level"), relevance=s.get("relevance"), facts=facts, status="consensus") async def _mirror_question_pattern_db(topic: str, pattern: dict) -> None: """Mirror question patterns {block title: [{subblock, question}]} into the DB table question_pattern.""" await db.delete_question_pattern(topic) for btitle, eintraege in pattern.items(): bnorm = _norm_title(btitle) if not bnorm or not isinstance(eintraege, list): continue for e in eintraege: if not isinstance(e, dict): continue sub = str(e.get("subblock", "")).strip() sn = _norm_title(sub) question = str(e.get("question", "")).strip() if not (sn and question): continue await db.upsert_question_pattern(topic, bnorm, sn, btitle, sub, question) async def _reset_db_from_phase(topic: str, label: str) -> None: """Discard DB content of phases ≥ `label` (canonical order Source…Artefacts).""" idx = _phase_idx(label) if idx <= 8: # Artefacts (flashcards/examples) await db.delete_sub_artefakte(topic) if idx <= 7: # Questions await db.delete_question_pattern(topic) if idx <= 6: # Outline await db.delete_outline(topic) if idx <= 2: # Subblocks (facts/levels/relevance go through sidecar→mirror) await db.delete_subblocks(topic) if idx <= 1: # Inventory: inventory + research steps — triage stays await db.delete_blocks(topic) await db.delete_pipeline_state(topic, ["Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"]) if idx <= 0: # Source: redo triage (coverage/content + step) await db.delete_coverage(topic) await db.delete_pipeline_state(topic, ["Source prep"]) async def generate_blocks(topic: str, instructions: str = "", provider: str = DEFAULT_PROVIDER, ab_phase: int | None = None, ab_step: int | None = None, to_step: int | None = None) -> None: if topic in _blocks_progress: return _blocks_progress[topic] = "Waiting…" _blocks_errors.pop(topic, None) files = _blocks_files(topic) final_path = files["final"] q = load_source(topic) folder = source_folder(topic) # projekt/uni/link → folder, thema → None instructions = q.get("spec") or instructions # prefer the persisted specification (also on resume) def set_p(msg: str, step: int | None = None) -> None: _blocks_progress[topic] = msg if step is not None: _blocks_step[topic] = step def is_cancelled() -> bool: return topic in _blocks_cancelled def aborted() -> None: _blocks_errors[topic] = "Cancelled — progress is preserved" _step_list = _blocks_steps(topic) def _past_limit(step: str) -> bool: # optional end limit: stop before any step past to_step return to_step is not None and step in _step_list and _step_list.index(step) > to_step ctx = GenContext(topic=topic, provider=provider, is_cancelled=is_cancelled) try: async with _semaphore: files["arbeit"].mkdir(parents=True, exist_ok=True) # Re-run from the chosen phase: delete artefacts from there; the fresh-start block # below is skipped (with a preserved sidecar it would otherwise wipe everything). if ab_step is not None: # fine sub-step re-run (takes precedence over ab_phase) await _reset_from_step(topic, ab_step) elif ab_phase is not None: phasen = _phases(topic) label = phasen[ab_phase - 1][0] if 1 <= ab_phase <= len(phasen) else "Inventory" _reset_from_phase(topic, label) await _reset_db_from_phase(topic, label) # A stage returning False ends generation; if it was a cancel, mark aborted first. async def _stage(coro) -> bool: ok = await coro if not ok and is_cancelled(): aborted() return ok # Step "Source prep": crawl (link) + PDFs + content/noise triage. if not await _stage(_prepare_source(ctx, set_p, files, q, folder, instructions)): return # "Create new": ONLY if truly everything is done (blocks.md AND # sidecar) → complete fresh start. If blocks.md exists without the sidecar, # it's a partial state (block B/C open) → resume, don't wipe. # On an explicit re-run (ab_phase) _reset_ab_phase already handled that. done = ab_phase is None and ab_step is None and final_path.exists() and _sidecar_schema(_json_file(files["sidecar"])) is not None if done: for p_old in _all_slot_files(files): p_old.unlink(missing_ok=True) await db.delete_pipeline_state(topic) await db.delete_blocks(topic) await db.delete_subblocks(topic) await db.delete_question_pattern(topic) await db.delete_coverage(topic) await db.delete_outline(topic) await db.delete_sub_artefakte(topic) # Inventory (DB): research loop → consolidation → clarification. if _past_limit("Research"): return if not await _stage(_research_batch(ctx, set_p, files, q, folder, instructions)): return if _past_limit("Consolidation"): return if not await _stage(_consolidate(ctx, set_p, files)): return if _past_limit("Clarification"): return if not await _stage(_clarify_inventory(ctx, set_p, files)): return if _past_limit("Dedup"): return if not await _stage(_dedup_inventory(ctx, set_p, files)): return if _past_limit("Blocks-Filter"): return if not await _stage(_filter_inventory(ctx, set_p, files)): return consensus_rows = await db.list_blocks(topic, status="consensus") entries = { i: (f"{b['title']} — {b['description']}" if b["description"] else b["title"]) for i, b in enumerate(consensus_rows, 1) } # Projects only: subject-field supplement — script/project is an excerpt, # a web agent adds canonically missing blocks, marked with [Supplement]. if q["type"] == "projekt" and not _past_limit("Supplement"): set_p("Supplementing subject field…", step=_step_idx(topic, "Supplement")) supp_path = files["ergaenzung"] supplements = _supplement_schema(_json_file(supp_path)) if supplements is None: supp_path.unlink(missing_ok=True) status, supplements = await run_single_slot( ctx, "Supplement", key=f"blocks-{topic}-ergaenzung-1", prompt=_prompt( "Blocks-Supplement", topic=topic, blocks="\n".join(f"- {t}" for t in entries.values()), out_path=supp_path, extra=_extra(instructions), ), role="quick", capabilities="full", payload=lambda result: _supplement_schema(_json_file(supp_path)), timeout=_timeout("ergaenzung"), ) if status == CANCELLED: aborted() return if status == FAILED: _blocks_errors[topic] = "Supplement failed (no valid result)" return idx = _title_index(entries) new = [(t, b) for t, b in supplements if _resolve_title(idx, t) is None] if new: _log(topic, f"Supplement: {len(new)} block(s) added from the subject field") start = max(entries, default=0) + 1 for off, (t, b) in enumerate(new): entries[start + off] = f"{t} — {b} [Supplement]" # Make titles unique and write the unsorted inventory entries = _unique_title(entries) atomic_write_text(final_path, "\n".join(f"{i}. {t}" for i, t in entries.items()) + "\n") if _past_limit("Subblocks find"): return # end limit inside the inventory → stop with blocks.md written # Block B + C: subblocks per block + levels → sidecar subblocks.json. # Non-destructive: blocks.md already exists; if the sidecar is missing, only # this part is retried on the next run. The guide falls back without the sidecar. if _sidecar_schema(_json_file(files["sidecar"])) is None: raw = _sub_raw_schema(_json_file(files["sub_roh"])) if raw is None: raw = await _subblocks_block(ctx, set_p, files, entries, instructions) if is_cancelled(): aborted() return if raw is None: return # error is set atomic_write_json(files["sub_roh"], raw, indent=1) if _past_limit("Facts find"): return # end limit after subblocks # Facts per sub (BEFORE the level): extract + verify source facts → facts.json. # Extract-once grounding — level/relevance/questions/guide feed on it. if not _facts_complete(files): res = await _facts_block(ctx, set_p, files, raw, q, folder, instructions) if is_cancelled(): aborted() return if res is None: return # error is set facts_map, discarded = res # Strike discarded (unsupportable) subs from raw — FIRST (resume-robust), then # facts.json. This way levels/relevance/outline/questions/guide no longer see them. if discarded: for bt, sns in discarded.items(): if bt in raw: raw[bt] = [s for s in raw[bt] if _norm_title(s) not in sns] raw = {bt: subs for bt, subs in raw.items() if subs} # drop empty blocks (_sub_roh_schema requires ≥1) atomic_write_json(files["sub_roh"], raw, indent=1) atomic_write_json(files["facts"], facts_map, indent=1) if _past_limit("Levels find"): return # end limit after facts (sidecar not yet valid → no DB mirror) sidecar = await _levels_block(ctx, set_p, files, raw, instructions) if is_cancelled(): aborted() return if sidecar is None: return # Merge facts into the sidecar subs (DB mirror + guide use). facts_map = _json_file(files["facts"]) if isinstance(facts_map, dict): for btitle, subs in sidecar.items(): fm = facts_map.get(btitle, {}) for sub in subs: if (fk := fm.get(_norm_title(sub["title"]))): sub["facts"] = fk atomic_write_json(files["sidecar"], sidecar, indent=1) # Block D: relevance per subblock (relevant/peripheral) → merge into the sidecar. # Own phase after the levels; drives the ProGuide format (all blocks # with ≥1 relevant subblock) and filters peripheral subs out of the guides. sidecar = _json_file(files["sidecar"]) if not _past_limit("Relevance find") and _sidecar_schema(sidecar) is not None and not _relevance_complete(sidecar): relevance_by_id = await _relevance_block(ctx, set_p, files, sidecar, instructions) if is_cancelled(): aborted() return if relevance_by_id is None: return # error is set gid = 0 for subs in sidecar.values(): for sub in subs: gid += 1 sub["relevance"] = relevance_by_id.get(gid, "relevant") atomic_write_json(files["sidecar"], sidecar, indent=1) # Block D.5: outline (blocks artifact) — chapter structure over ALL blocks, # only read by the guide. Format-agnostic; the guide filters per format. if not _past_limit("Outline") and not _outline_complete(files): await _outline_block(ctx, set_p, files, entries, instructions) if is_cancelled(): aborted() return # Block E: question pattern per relevant subblock × type → own sidecar. # At exam time each agent draws a pattern without replacement and formulates # a question from it — distinct seeding prevents the duplicate questions of live generation. sidecar = _json_file(files["sidecar"]) if not _past_limit("Questions find") and _sidecar_schema(sidecar) is not None and _relevance_complete(sidecar) and not _question_pattern_complete(topic): pattern = await _question_pattern_block(ctx, set_p, files, sidecar, instructions) if is_cancelled(): aborted() return if pattern is None: return # cancel atomic_write_json(files["question_pattern"], pattern, indent=1) # Block F: learning artefacts (flashcards/examples) from the facts — bonus, # presented by the frontend. Does not abort the run (artefacts are optional). sidecar = _json_file(files["sidecar"]) if not _past_limit("Flashcards") and _sidecar_schema(sidecar) is not None and not _artefacts_complete(files): artefacts = await _artefacts_block(ctx, set_p, files, sidecar, instructions) if is_cancelled(): aborted() return if artefacts is None: return # cancel (error/cancel) # DB mirror (bridge): write the final sidecar + question-pattern state into the DB. sidecar = _json_file(files["sidecar"]) if _sidecar_schema(sidecar) is not None: await _mirror_sidecar_db(topic, sidecar) pattern = _json_file(files["question_pattern"]) if isinstance(pattern, dict) and pattern: await _mirror_question_pattern_db(topic, pattern) # Outline (title-based, robust against number drift) → DB. plan = _json_file(files["outline"]) if isinstance(plan, dict) and plan.get("chapters"): kapitel = [ {"title": ch.get("title", "Chapter"), "blocks": [_title(entries[n]) for n in ch.get("numbers", []) if n in entries]} for ch in plan["chapters"] ] await db.set_outline(topic, json.dumps({"chapters": kapitel}, ensure_ascii=False)) # Artefacts → DB (flashcard/example per sub, diagram per block). artefacts = _json_file(files["artefakte"]) if isinstance(artefacts, dict) and _sidecar_schema(sidecar) is not None: await _mirror_artefacts_db(topic, sidecar, artefacts) except Exception as e: log.exception("[%s] Blocks generation failed", topic) _blocks_errors[topic] = str(e)[:2000] finally: # No file cleanup: intermediate files stay for resume / traceability. _blocks_progress.pop(topic, None) _blocks_step.pop(topic, None) _blocks_cancelled.discard(topic) clear_scope(f"blocks-{topic}-") # clear the scope → restart isn't blocked