update

.env.example

@@ -5,3 +5,9 @@ CLAUDE_CODE_OAUTH_TOKEN=
 
 # MiniMax-Provider: API-Key aus der MiniMax-Console (Coding-Plan).
 MINIMAX_API_KEY=
+
+# Agent-Parallelität (optional). Zwei verschachtelte Limits, Default je 10 = bisheriges Verhalten.
+# Global gilt über ALLE Themen, der Thema-Wert je Thema. Lokal das globale Limit hochsetzen,
+# um mehrere Themen parallel mit je 10 Agenten zu fahren (z.B. global 50, Thema 10).
+# MAX_CONCURRENT_AGENTS=10
+# MAX_CONCURRENT_AGENTS_PER_TOPIC=10

backend/agents.py

@@ -13,9 +13,11 @@ import signal
 import tempfile
 import time
 import urllib.request
+from contextlib import asynccontextmanager
 from pathlib import Path
 
-from config import PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS, MAX_CONCURRENT_INTERACTIVE
+from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS,
+                    MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_INTERACTIVE)
 
 log = logging.getLogger("creator.agents")
 
@@ -43,6 +45,24 @@ def _scope_cancelled(agent_key: str) -> bool:
 # does not count against the agent timeout.
 _batch_sem = asyncio.Semaphore(MAX_CONCURRENT_AGENTS)
 _interactive_sem = asyncio.Semaphore(MAX_CONCURRENT_INTERACTIVE)
+# Per-topic caps (lazily created): each topic gets its own batch semaphore of size
+# MAX_CONCURRENT_AGENTS_PER_TOPIC, nested INSIDE the global _batch_sem.
+_topic_sems: dict[str, asyncio.Semaphore] = {}
+
+
+@asynccontextmanager
+async def _batch_gate(scope: str | None):
+    """Acquire a batch slot: per-topic semaphore FIRST, then the global one. The order matters —
+    a waiter holds only its (per-topic) slot while queueing for the global cap, so a saturated topic
+    never blocks other topics on the global semaphore. scope=None → global cap only."""
+    topic_sem = _topic_sems.setdefault(scope, asyncio.Semaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None
+    if topic_sem is None:
+        async with _batch_sem:
+            yield
+    else:
+        async with topic_sem:
+            async with _batch_sem:
+                yield
 
 # Serialize OpenCode starts: processes starting simultaneously collide on the
 # internal session DB ("database is locked", exit after <1s). The short
@@ -116,6 +136,7 @@ async def run_agent(
     role: str = "fast",
     capabilities: str = "none",
     lane: str = "batch",
+    scope: str | None = None,
 ) -> tuple[int, str, str]:
     if _scope_cancelled(agent_key):  # before queueing: don't even enter the queue
         return 1, "", "cancelled"
@@ -123,8 +144,8 @@ async def run_agent(
         return 1, "", f"Unknown provider: {provider}"
     if shutil.which(PROVIDERS[provider]["cli"]) is None:
         return 1, "", f"CLI '{PROVIDERS[provider]['cli']}' not installed (provider: {provider})"
-    sem = _interactive_sem if lane == "interactive" else _batch_sem
-    async with sem:
+    gate = _interactive_sem if lane == "interactive" else _batch_gate(scope)
+    async with gate:
         if _scope_cancelled(agent_key):  # after the acquire: cancelled in the queue → no spawn
             return 1, "", "cancelled"
         if PROVIDERS[provider]["cli"] == "opencode":

backend/blocks.py

@@ -23,7 +23,7 @@ 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 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, EMBEDDING_SUB_SAME
 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
@@ -198,7 +198,7 @@ def _blocks_steps(topic: str) -> tuple:
     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")
+    base = ("Research", "Consolidation", "Clarification", "Blocks-Filter")
     rest = (
         "Subblocks find", "Subblocks select", "Subblocks clarify",
         "Facts find", "Facts check", "Facts fix",
@@ -228,7 +228,7 @@ def _report_p(set_p, topic: str, step: str):
 # Special steps (Source laden, Supplement) belong to the "Inventory" phase.
 PHASEN = (
     ("Source", ("Source prep",)),
-    ("Inventory", ("Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter", "Supplement")),
+    ("Inventory", ("Research", "Consolidation", "Clarification", "Blocks-Filter", "Supplement")),
     ("Subblocks", ("Subblocks find", "Subblocks select", "Subblocks clarify")),
     ("Facts", ("Facts find", "Facts check", "Facts fix")),
     ("Levels", ("Levels find", "Levels select", "Levels clarify")),
@@ -317,7 +317,7 @@ async def _resume_step(topic: str) -> int:
     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"):
+        for step in ("Source prep", "Research", "Consolidation", "Clarification", "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
@@ -451,15 +451,19 @@ def _reset_from_phase(topic: str, label: str) -> None:
         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)."""
+async def _reset_from_step(topic: str, step_idx: int, to_idx: int | None = None) -> None:
+    """Fine reset FROM a sub-step (0-based index in _blocks_steps). With `to_idx` set, ONLY the
+    span [step_idx, to_idx] is reset — later steps and the blocks.md aggregate stay (isolated
+    single-step / bounded-range regenerate). Without it the reset cascades to the end (full re-run).
+    Earlier steps always stay. Inventory sub-steps reconstruct the DB status from the artefacts
+    (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:])
+    last = len(fine_steps) - 1 if to_idx is None else max(step_idx, min(to_idx, len(fine_steps) - 1))
+    span = fine_steps[step_idx:last + 1]
+    affected = set(span)
     files = _blocks_files(topic)
     work_dir = files["arbeit"]
 
@@ -468,7 +472,7 @@ async def _reset_from_step(topic: str, step_idx: int) -> None:
             for p in work_dir.glob(pat):
                 p.unlink(missing_ok=True)
 
-    await db.delete_pipeline_state(topic, list(fine_steps[step_idx:]))
+    await db.delete_pipeline_state(topic, span)
     # 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)
@@ -501,21 +505,12 @@ async def _reset_from_step(topic: str, step_idx: int) -> None:
     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):
+    if "Blocks-Filter" in affected and not ({"Clarification", "Consolidation", "Research"} & 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.
@@ -527,8 +522,9 @@ async def _reset_from_step(topic: str, step_idx: int) -> None:
         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:
+    # statuses of the kept earlier steps let those skip). On a BOUNDED reset (to_idx set) the later steps
+    # are kept on purpose → keep their aggregate too.
+    if to_idx is None and {"Research", "Consolidation", "Clarification", "Blocks-Filter"} & affected:
         files["final"].unlink(missing_ok=True)
 
 
@@ -867,7 +863,7 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i
 
         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]
+        pending = [(j, p) for j, p in enumerate(paths, 1) if not _parse_subblocks(_read(p))]  # {} (empty/missing) also pending — _parse_subblocks never returns None
         for _, p in pending:
             p.unlink(missing_ok=True)
         if pending:
@@ -890,12 +886,13 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i
             atomic_write_text(fp, text)
             return
 
-        # code majority per block/sub-norm: keep if a majority of judges list it (tie → keep).
+        # Semantic majority per block: cluster the judges' subblocks by meaning, keep a cluster a
+        # majority of judges contributed to. Exact-string voting discarded paraphrased core facts
+        # (each judge wording → 1 vote) while verbatim side-notes won.
         block_texts_out = []
         for num in chunk:
-            votes: dict[str, int] = {}
-            form: dict[str, str] = {}
-            for d in outs:
+            cand: list[tuple[int, str]] = []  # (judge index, subblock) — exact-dedup per judge
+            for ji, d in enumerate(outs):
                 seen = set()
                 for marker, subs in d.items():
                     if _resolve_title(chunk_idx, marker) != num:
@@ -905,9 +902,10 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i
                         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)]
+                        cand.append((ji, sub))
+            if not cand:
+                continue
+            kept = await _cluster_vote(cand, len(outs))
             if kept:
                 block_texts_out.append(f"<!-- block: {title_by_num[num]} -->\n" + "\n".join(f"- {s}" for s in kept))
         atomic_write_text(fp, "\n\n".join(block_texts_out))
@@ -947,6 +945,42 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i
     return raw
 
 
+async def _cluster_vote(cand: list[tuple[int, str]], n_judges: int) -> list[str]:
+    """Semantic majority over the judges' subblocks. Cluster by cosine ≥ EMBEDDING_SUB_SAME (same
+    point), keep a cluster a majority of distinct judges contributed to, pick the longest (most
+    informative) phrasing. Replaces exact-string voting, which split paraphrases of the same fact.
+    Model missing → norm-union fallback (keep everything ≥1 judge, exact-dedup — never lose content)."""
+    texts = [s for _, s in cand]
+    sims = (await asyncio.to_thread(embedding.embed_sims, texts)
+            if EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available) else None)
+    if sims is None:
+        out, seen = [], set()
+        for _, s in cand:
+            sn = _norm_title(s)
+            if sn and sn not in seen:
+                seen.add(sn)
+                out.append(s)
+        return out
+    parent = list(range(len(texts)))
+    def find(x: int) -> int:
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+    for i in range(len(texts)):
+        for k in range(i + 1, len(texts)):
+            if float(sims[i][k]) >= EMBEDDING_SUB_SAME:
+                parent[find(i)] = find(k)
+    clusters: dict[int, list[int]] = {}
+    for i in range(len(texts)):
+        clusters.setdefault(find(i), []).append(i)
+    kept = []
+    for idxs in clusters.values():
+        if len({cand[i][0] for i in idxs}) * 2 >= n_judges:        # majority of judges
+            kept.append(max((texts[i] for i in idxs), key=len))    # longest phrasing
+    return kept
+
+
 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
@@ -1324,7 +1358,7 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst
         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)
+                _timeout("content_check", len(per)), provider=provider, role="judge", capabilities=caps, scope=topic)
             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] = {}
@@ -1920,6 +1954,37 @@ async def _prepare_source(ctx: GenContext, set_p, files: dict, q: dict, folder,
     return True
 
 
+async def _ingest_research(topic: str, reader_id: str, text: str) -> None:
+    """Parse one reader's research file (lines `title — desc — source`) into block candidates and
+    upsert them with the reader id (one reader = one vote per concept). Shared by the research step
+    and the candidate re-ingest below."""
+    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)
+
+
+async def _reingest_research_files(topic: str, work_dir: Path) -> None:
+    """Rebuild the research candidates in the blocks DB from the saved research-*.md reader files
+    (no agents, no web calls). Consolidation CONSUMES the candidates (overwrites them with
+    consensus/rest), so a consolidation re-run / reset needs them restored. Reader id = file suffix
+    (pure number N → "tN" for the web readers; "a…"/"b…" section/batch readers stay as-is)."""
+    for p in sorted(work_dir.glob("research-*.md")):
+        suffix = p.stem[len("research-"):]
+        if not suffix:
+            continue
+        rid = f"t{suffix}" if suffix.isdigit() else suffix
+        if text := _file_payload(p):
+            await _ingest_research(topic, rid, text)
+
+
 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
@@ -1933,17 +1998,7 @@ async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder,
     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)
+        await _ingest_research(topic, reader_id, text)
 
     pages = await db.list_content(topic)  # pages marked as content by the triage
     if not pages and folder:
@@ -2068,31 +2123,6 @@ async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder,
     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")
@@ -2146,47 +2176,58 @@ def _canonical(candidates: list[dict], idxs: list[int], seen_norm: set[str]) ->
     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)."""
+async def _pairwise_groups(ctx: GenContext, set_p, work_dir: Path, candidates: list[dict],
+                               blocks: list[list[int]], sims) -> list[list[int]] | None:
+    """Verify candidate PAIRS individually (ja/nein) inside each similarity block, then form
+    COMPLETE-LINK cliques — same entity-resolution mechanism as the dedup pass: no chaining
+    (A=B + B=C without A=C does NOT merge), no aspect over-merging like the old N→groups judge.
+    Only block-internal pairs with cosine ≥ DEDUP_PAIR_FLOOR are checked; members without a
+    confirmed edge stay singletons. → final groups (global candidate indices) · None on cancel."""
     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
+    n = len(candidates)
+    pairs: list[tuple[int, int]] = []  # block-internal candidate pairs above the pair floor
+    for b in blocks:
+        for x in range(len(b)):
+            for y in range(x + 1, len(b)):
+                i, j = b[x], b[y]
+                if float(sims[i][j]) >= DEDUP_PAIR_FLOOR:
+                    pairs.append((i, j))
+    if not pairs:
+        return [[i] for i in range(n)]
+    packages = [pairs[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(pairs), DEDUP_PAIRS_CHUNK)]
 
-    def _line(k: int, g: int) -> str:
-        b = candidates[g]
-        return f"{k}. {b['title']}" + (f" — {b['description']}" if b["description"] else "")
+    def pair_path(pi): return work_dir / f"consolidation-paar-c{pi}.json"
 
-    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)
+    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: {candidates[a]['title']} — {candidates[a]['description']}"
+            f"\nB: {candidates[b]['title']} — {candidates[b]['description']}"
+            for j, (a, b) in enumerate(paare))
+        await run_single_slot(
+            ctx, f"Consolidation pairs {pi}",
+            key=f"blocks-{topic}-consolidation-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([_grp(bi, b) for bi, b in multi],
-                              len(multi), _report_p(set_p, topic, step))
-    return outcome
+    await _gather_progress([_filt(pi, p) for pi, p in enumerate(packages)],
+                              len(packages), _report_p(set_p, topic, "Consolidation"))
+    if is_cancelled():
+        return None
+    edge_list: list[tuple[int, int]] = []
+    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))
+    cliques = _cliques(n, edge_list)
+    covered = {i for g in cliques for i in g}
+    return cliques + [[i] for i in range(n) if i not in covered]
 
 
 async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates: list[dict]) -> bool:
@@ -2198,11 +2239,12 @@ async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, 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).
+    # Level 1: coarse similarity blocks (capped, no giant component) — pure blocking for recall.
     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():
+    # Level 2: verify candidate PAIRS individually + complete-link cliques (no chaining, no aspect
+    # over-merging) instead of an N→groups judge that fused whole topics into one block.
+    groups = await _pairwise_groups(ctx, set_p, work_dir, candidates, blocks, sims)
+    if groups is None or is_cancelled():
         return False
 
     def _min_cos(idxs):  # internal coherence as a check (chains would be ~0.3)
@@ -2230,7 +2272,7 @@ async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates
                       "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) "
+    _log(topic, f"Consolidation (pairwise): {len(blocks)} blocks ({multi_blocks} multi) "
                 f"→ {len(groups)} clusters from {len(candidates)} candidates "
                 f"→ {len(consensus)} consensus / {len(rest)} rest")
 
@@ -2251,6 +2293,11 @@ async def _consolidate(ctx: GenContext, set_p, files: dict) -> bool:
         return True
     set_p("Consolidating research…", step=_step_idx(topic, "Consolidation"))
     candidates = await db.list_blocks(topic)
+    if not candidates:
+        # Candidates were consumed by an earlier consolidation (overwritten with consensus/rest) or
+        # wiped by a reset → rebuild them from the saved research files so this step can re-run.
+        await _reingest_research_files(topic, files["arbeit"])
+        candidates = await db.list_blocks(topic)
     if not candidates:
         _blocks_errors[topic] = "Consolidation: no candidates"
         return False
@@ -2434,7 +2481,7 @@ async def _clarify_inventory(ctx: GenContext, set_p, files: dict) -> bool:
                     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")}
+        seen_norm = {b["title_norm"] for b in await db.list_blocks(topic)}  # all stati: UNIQUE(topic,title_norm) spans every status, not just consensus
         for b in check_rows:
             accept = votes.get(b["title_norm"], 0) * 2 >= len(outs)
             if not accept:
@@ -2494,83 +2541,6 @@ def _cliques(n: int, edge_list: list[tuple[int, int]]) -> list[list[int]]:
     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."""
@@ -2963,7 +2933,7 @@ async def _artefacts_block(ctx: GenContext, set_p, files: dict, sidecar: dict, i
             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)
+                    _timeout("content_check", len(items)), provider=provider, role="judge", capabilities=caps, scope=topic)
                 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:
@@ -3073,7 +3043,7 @@ async def _reset_db_from_phase(topic: str, label: str) -> None:
         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"])
+        await db.delete_pipeline_state(topic, ["Research", "Consolidation", "Clarification", "Blocks-Filter"])
     if idx <= 0:  # Source: redo triage (coverage/content + step)
         await db.delete_coverage(topic)
         await db.delete_pipeline_state(topic, ["Source prep"])
@@ -3114,7 +3084,7 @@ async def generate_blocks(topic: str, instructions: str = "", provider: str = DE
             # 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)
+                await _reset_from_step(topic, ab_step, to_step)
             elif ab_phase is not None:
                 phasen = _phases(topic)
                 label = phasen[ab_phase - 1][0] if 1 <= ab_phase <= len(phasen) else "Inventory"
@@ -3156,9 +3126,6 @@ async def generate_blocks(topic: str, instructions: str = "", provider: str = DE
             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

backend/config.py

@@ -1,3 +1,4 @@
+import os
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
@@ -41,11 +42,17 @@ EMBEDDING_BLOCK_CAP = 25      # max. titles per block (keep the LLM list short/s
 # block context — from this cosine on two are the same statement (checked on aak: ≥0.88 are
 # without exception true duplicates). Conservative 0.90 so different aspects (∈NP ≠ NP-hard) stay separate.
 EMBEDDING_SUB_DUP = 0.90
+# Cosine: two judge-subblocks state the SAME point → one cluster in the clarify majority vote.
+# Lower than _DUP because it must merge paraphrases (not just typo-variants). Empirically 0.80 keeps
+# distinct aspects (∈NP vs NP-hard) apart while clustering re-wordings of the same fact.
+EMBEDDING_SUB_SAME = 0.80
 
-# Cap for concurrent CLI agent processes (across all generations).
-# Own lane for interactive calls (chat, elements) so they don't hang behind
-# running writers in the queue.
-MAX_CONCURRENT_AGENTS = 10
+# Caps for concurrent CLI agent processes (env-overridable). Two nested limits, both always active:
+# a per-topic cap and a global cap across all topics. Defaults 10/10 = previous behavior (global
+# dominates). Locally raise the global cap to actually parallelize across topics (per-topic stays 10).
+# Own lane for interactive calls (chat, elements) so they don't hang behind running writers.
+MAX_CONCURRENT_AGENTS = int(os.getenv("MAX_CONCURRENT_AGENTS", "10"))                      # global, all topics
+MAX_CONCURRENT_AGENTS_PER_TOPIC = int(os.getenv("MAX_CONCURRENT_AGENTS_PER_TOPIC", "10"))  # per topic
 MAX_CONCURRENT_INTERACTIVE = 8
 
 # Grace window of the consensus races (blocks, guide, OnePager): after the first

backend/guide.py

@@ -469,7 +469,7 @@ async def _generate_sections(
                     topic=topic, assignment=assignments[i], facts=facts,
                     out_path=content_paths[i], extra=_extra(instructions),
                 ),
-                _timeout("content", chunk_sizes[i]), provider=provider, role="guide", capabilities="full",
+                _timeout("content", chunk_sizes[i]), provider=provider, role="guide", capabilities="full", scope=topic,
             )
             for i in pending
         ], writer_count, report, start=writer_count - len(pending))
@@ -509,7 +509,7 @@ async def _generate_sections(
                         topic=topic, assignment=_assignment_subs(followup_chunks[k], entries, subs_by_title),
                         facts=facts, out_path=followup_paths[k], extra=_extra(instructions),
                     ),
-                    _timeout("content", len(followup_chunks[k][0]["nums"])), provider=provider, role="guide", capabilities="full",
+                    _timeout("content", len(followup_chunks[k][0]["nums"])), provider=provider, role="guide", capabilities="full", scope=topic,
                 )
                 for k in followup_pending
             ], len(followup_chunks), report_n, start=len(followup_chunks) - len(followup_pending))
@@ -583,7 +583,7 @@ async def _generate_sections(
                     ),
                     out_path=fix_paths[i], extra=_extra(instructions),
                 ),
-                _timeout("content", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full",
+                _timeout("content", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", scope=topic,
             )
             for i in fix_pending
         ], return_exceptions=True)
@@ -620,7 +620,7 @@ async def _generate_sections(
                     contents=content_text(w_chunks[i]),
                     spec=spec, out_path=paths[i], extra=_extra(instructions),
                 ),
-                _timeout("writer", 1), provider=provider, role="guide", capabilities="files",
+                _timeout("writer", 1), provider=provider, role="guide", capabilities="files", scope=topic,
             )
             for i in pending
         ], len(w_chunks), report, start=len(w_chunks) - len(pending))
@@ -668,7 +668,7 @@ async def _generate_sections(
                         topic=topic, format_name=format_name, assignment=_assignment_subs(nw_chunks[k], entries, subs_by_title),
                         contents=content_text(nw_chunks[k]), spec=spec, out_path=nw_paths[k], extra=_extra(instructions),
                     ),
-                    _timeout("writer", 1), provider=provider, role="guide", capabilities="files",
+                    _timeout("writer", 1), provider=provider, role="guide", capabilities="files", scope=topic,
                 )
                 for k in nw_pending
             ], len(nw_chunks), report_nw, start=len(nw_chunks) - len(nw_pending))
@@ -770,7 +770,7 @@ async def _generate_sections(
                     tasks=tasks_text(fix_chunks[i], problems_by_num),
                     out_path=fix_paths[i], extra=_extra(instructions),
                 ),
-                _timeout("writer", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full",
+                _timeout("writer", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", scope=topic,
             )
             for i in fix_pending
         ], return_exceptions=True)

backend/pipeline.py

@@ -191,6 +191,7 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout:
         task = asyncio.create_task(run_agent(
             slot["key"], slot["prompt"], timeout,
             provider=provider, role=slot["role"], capabilities=slot["capabilities"],
+            scope=topic,
         ))
         tasks[task] = i
 

frontend/src/components/BlocksOverview.vue

@@ -47,7 +47,11 @@ function arm(action, fn) {
   if (confirm.value === action) { confirm.value = null; fn() }
   else confirm.value = action
 }
-function regenerateFromHere() { const from = startSel.value, to = endSel.value; clearSel(); emit('restartFrom', { from, to }) }
+// regenerate: re-run ONLY the picked step(s). No end → bound to the start itself (single step);
+// with an end → the whole [start, end] range. Later steps stay intact (bounded reset).
+function regenerateFromHere() { const from = startSel.value, to = endSel.value ?? startSel.value; clearSel(); emit('restartFrom', { from, to }) }
+// continue: run from the start point straight through to the end of the pipeline (full cascade).
+function continueFromHere() { const from = startSel.value; clearSel(); emit('restartFrom', { from, to: null }) }
 function deleteFromHere() { const from = startSel.value; clearSel(); emit('resetFrom', from) }
 
 const items = ref([])
@@ -139,7 +143,8 @@ const subTotal = computed(() => items.value.reduce((n, b) => n + (b.subblocks?.l
       </div>
       <div v-if="startSel !== null && !generating" class="bk-step-actions">
         <span class="bk-step-actions-label">From «{{ startLabel }}»<span v-if="endSel !== null"> to «{{ endLabel }}»</span>:</span>
-        <button class="bk-act play" @click="regenerateFromHere">↻ regenerate</button>
+        <button class="bk-act play" @click="regenerateFromHere" :title="endSel === null ? 'Re-run only this step; later steps stay' : 'Re-run every step in the range; later steps stay'">↻ regenerate {{ endSel === null ? 'step' : 'range' }}</button>
+        <button v-if="endSel === null" class="bk-act" @click="continueFromHere" title="Run from here through to the end">▶ continue</button>
         <button class="bk-act danger" :class="{ armed: confirm === 'reset' }" @click="arm('reset', deleteFromHere)">{{ confirm === 'reset' ? 'Sure?' : '✕ delete all' }}</button>
         <button class="bk-act ghost" @click="clearSel">Cancel</button>
       </div>

templates/Prompt/Blocks-Paar-Filter.md

@@ -1,18 +1,23 @@
-Two research passes have noted blocks for the topic "{topic}". For EACH pair, decide whether A and B denote the SAME block (the same concept, just worded differently) → **ja**, or whether they are TWO DIFFERENT blocks → **nein**.
+Research has noted block candidates for the topic "{topic}". For EACH pair, decide whether A and B denote the SAME block (the same concept, just worded differently) → **ja**, or whether they are TWO DIFFERENT blocks → **nein**.
 
 PAIRS:
 {pairs}
 
 Rules:
-- **Watch the CORE ENTITY first** (the problem/object in question): Clique, Vertex Cover, Independent Set, Dominating Set, Set Cover, FVS, Knapsack … If the entities are DIFFERENT → **nein**, no matter how identical the phrasing.
-- Identical phrasing is deceptive. These pairs are **nein** (different entity despite nearly identical wording):
-  - "Lower Bound **Clique** bzgl. Knoten" ↔ "Lower Bound **Vertex Cover** bzgl. Knoten"
-  - "Lower Bound Clique bzgl. **Knoten**" ↔ "Lower Bound Clique bzgl. **Kanten**"
-  - "Verifizierer für **FVS**" ↔ "Verifizierer für **Knapsack**"
-  - "**Cliquenproblem**" ↔ "**Vertex-Cover-Problem**"
-- **ja** only on genuine semantic equivalence: same solution to the same problem, the same entity, just different wording/naming (e.g. "SET COVER" ↔ "Mengenüberdeckungsproblem", "Cliquenproblem" ↔ "k-CLIQUE", "List Scheduling" ↔ "LPT-Algorithmus").
-- **nein** also for different aspects of the same problem: "Set Cover (Problem)" ↔ "Set Cover ETH-Schranke"; a problem ↔ its reduction to another; a problem ↔ its verifier.
-- When in doubt **nein** — better two separate blocks than wrongly merging two concepts.
+- **Watch the CORE ENTITY first** (the object/concept/operation in question). If the entities are DIFFERENT → **nein**, no matter how identical the phrasing. Examples of **nein** despite near-identical wording:
+  - "Plugin-**Lebenszyklus**" ↔ "App-**Lebenszyklus**" (different object: Plugin vs App)
+  - "**install()**-Methode" ↔ "**uninstall()**-Methode" (different operation)
+  - "Lower Bound **Clique**" ↔ "Lower Bound **Vertex Cover**" (different problem)
+  - "Present **Perfect**" ↔ "Past **Perfect**" (different tense)
+- **ja** ONLY on genuine semantic equivalence: the same concept/operation, just different wording, naming, language or abbreviation. Examples:
+  - "EntityRepository" ↔ "Repository für CRUD-Operationen am DAL"
+  - "Set Cover" ↔ "Mengenüberdeckungsproblem"
+  - "List Scheduling" ↔ "LPT-Algorithmus"
+- **nein** for different ASPECTS, properties, parts or methods of the same thing — they are their own blocks (a later step folds true fragments back in):
+  - a concept ↔ one of its properties/details ("DAL" ↔ "DAL-Versionierung")
+  - a concept ↔ a single method/step of it ("Plugin-Lebenszyklus" ↔ "install()-Methode")
+  - a thing ↔ its sub-component, its verifier, its reduction to another thing
+- When in doubt → **nein**. Better two separate blocks than wrongly merging two concepts.
 
 Write ONLY the JSON file to: {out_path}