From b5398f73d239bfdcc6c0bbb7de2fd8b7cad4f8fd Mon Sep 17 00:00:00 2001 From: team3 Date: Wed, 1 Jul 2026 22:01:32 +0200 Subject: [PATCH] update --- backend/agents.py | 131 +++- backend/blocks.py | 772 ++++++++------------- backend/config.py | 3 + backend/database.py | 195 ++++++ backend/kanban.py | 748 ++++++++++++++++++++ backend/models.py | 1 + backend/pipeline.py | 37 +- backend/routes.py | 24 +- frontend/src/App.vue | 6 +- frontend/src/api.js | 22 +- frontend/src/components/BlocksOverview.vue | 75 +- templates/Prompt/Blocks-Dependency.md | 17 + templates/Prompt/Blocks-Filter-Check.md | 23 + templates/Prompt/Blocks-Naming-Check.md | 15 + templates/Prompt/Blocks-Naming.md | 15 + templates/Prompt/Blocks-Research.md | 3 +- templates/Prompt/Blocks-Small.md | 15 + 17 files changed, 1580 insertions(+), 522 deletions(-) create mode 100644 backend/kanban.py create mode 100644 templates/Prompt/Blocks-Dependency.md create mode 100644 templates/Prompt/Blocks-Filter-Check.md create mode 100644 templates/Prompt/Blocks-Naming-Check.md create mode 100644 templates/Prompt/Blocks-Naming.md create mode 100644 templates/Prompt/Blocks-Small.md diff --git a/backend/agents.py b/backend/agents.py index d19eda2..8cbf261 100644 --- a/backend/agents.py +++ b/backend/agents.py @@ -5,6 +5,7 @@ respective provider fails — the other keeps running unchanged. """ import asyncio +import heapq import logging import os import re @@ -22,6 +23,17 @@ from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS, log = logging.getLogger("creator.agents") _active_processes: dict[str, asyncio.subprocess.Process] = {} +_active_started: dict[str, float] = {} # agent_key → wall-clock start (for the live runtime display) + + +def active_agents(scope_prefix: str | None = None) -> list[dict]: + """Currently running agents and how long they've been running. Filter by key prefix + (e.g. f"blocks-{topic}-") for one topic. → [{key, runtime}] sorted longest-first.""" + now = time.time() + out = [{"key": k, "runtime": round(now - t, 1)} + for k, t in list(_active_started.items()) + if k in _active_processes and (not scope_prefix or k.startswith(scope_prefix))] + return sorted(out, key=lambda a: -a["runtime"]) # Cancelled scopes (key prefixes, symmetric to kill_process). An agent whose # key starts with one of these prefixes aborts BEFORE the spawn — so agents WAITING @@ -43,26 +55,71 @@ def _scope_cancelled(agent_key: str) -> bool: # Caps the real CLI processes — independent of the pipeline semaphore in # generator.py. The acquire happens BEFORE the spawn so that queue wait time # does not count against the agent timeout. -_batch_sem = asyncio.Semaphore(MAX_CONCURRENT_AGENTS) +class _PrioritySemaphore: + """asyncio.Semaphore variant: when slots are scarce, the LOWEST priority number is served first + (FIFO within the same priority). Lets earlier pipeline columns grab agents before later ones.""" + def __init__(self, value: int): + self._value = value + self._waiters: list = [] # heap of [priority, seq, future] + self._seq = 0 + + async def acquire(self, priority: int = 100): + if self._value > 0: + self._value -= 1 + return + fut = asyncio.get_event_loop().create_future() + entry = [priority, self._seq, fut] + self._seq += 1 + heapq.heappush(self._waiters, entry) + try: + await fut # release() hands us the slot directly (no value change) + except BaseException: + entry[2] = None # tombstone so release() skips this dead waiter + if fut.done() and not fut.cancelled(): + self.release() # granted just before we were cancelled → pass it on + raise + + def release(self): + while self._waiters: + entry = heapq.heappop(self._waiters) + if entry[2] is not None and not entry[2].done(): + entry[2].set_result(None) # hand the slot straight to the highest-priority waiter + return + self._value += 1 + + +_batch_sem = _PrioritySemaphore(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] = {} +# Per-topic caps (lazily created): each topic gets its own priority semaphore of size +# MAX_CONCURRENT_AGENTS_PER_TOPIC, nested INSIDE the global _batch_sem. Priority-based too, so the +# per-topic queue can't undo the global priority when one topic is the only load. +_topic_sems: dict[str, _PrioritySemaphore] = {} + +# Earlier kanban columns get the scarce global slot first (smaller = higher priority). +_STAGE_PRIORITY = ("research", "verify", "naming", "small", "dep") + + +def _agent_priority(key: str) -> int: + for i, tag in enumerate(_STAGE_PRIORITY): + if f"-{tag}-" in key or key.endswith(f"-{tag}"): + return i + return len(_STAGE_PRIORITY) # downstream agents (subblocks/facts/…) after the inventory columns @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 +async def _batch_gate(scope: str | None, priority: int): + """Per-topic slot FIRST (fair), then the GLOBAL slot by priority (earlier columns win when + agents are scarce). Order matters — a waiter holds only its per-topic slot while queueing globally.""" + topic_sem = _topic_sems.setdefault(scope, _PrioritySemaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None + if topic_sem is not None: + await topic_sem.acquire(priority) + await _batch_sem.acquire(priority) + try: + yield + finally: + _batch_sem.release() + if topic_sem is not None: + topic_sem.release() # Serialize OpenCode starts: processes starting simultaneously collide on the # internal session DB ("database is locked", exit after <1s). The short @@ -122,6 +179,7 @@ def kill_process(agent_key_prefix: str) -> None: for key, process in list(_active_processes.items()): if process.returncode is not None: # clean up dead entries while iterating _active_processes.pop(key, None) + _active_started.pop(key, None) continue if key.startswith(agent_key_prefix): log.debug("kill agent %s", key) @@ -137,6 +195,7 @@ async def run_agent( capabilities: str = "none", lane: str = "batch", scope: str | None = None, + on_line=None, ) -> tuple[int, str, str]: if _scope_cancelled(agent_key): # before queueing: don't even enter the queue return 1, "", "cancelled" @@ -144,16 +203,16 @@ 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})" - gate = _interactive_sem if lane == "interactive" else _batch_gate(scope) + gate = _interactive_sem if lane == "interactive" else _batch_gate(scope, _agent_priority(agent_key)) 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": - return await _run_opencode(agent_key, prompt, timeout, provider, role, capabilities) + return await _run_opencode(agent_key, prompt, timeout, provider, role, capabilities, on_line=on_line) return await _run_claude_cli(agent_key, prompt, timeout, role, capabilities) -async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int, stagger: bool = False) -> tuple[int, str, str]: +async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int, stagger: bool = False, on_line=None) -> tuple[int, str, str]: start = time.monotonic() async def spawn(): @@ -172,12 +231,29 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, else: process = await spawn() _active_processes[agent_key] = process + _active_started[agent_key] = time.time() try: try: - stdout, stderr = await asyncio.wait_for( - process.communicate(input=stdin_data), - timeout=timeout, - ) + if on_line is not None: + # Streaming path: read stdout line by line, hand each raw line to on_line LIVE. + out_chunks: list[str] = [] + async def _pump(): + async for raw in process.stdout: + s = raw.decode("utf-8", errors="replace") + out_chunks.append(s) + try: + on_line(s) + except Exception: + log.debug("on_line callback failed", exc_info=True) + await asyncio.wait_for(_pump(), timeout=timeout) + await process.wait() + stderr_b = await process.stderr.read() + stdout, stderr = "".join(out_chunks).encode("utf-8"), stderr_b + else: + stdout, stderr = await asyncio.wait_for( + process.communicate(input=stdin_data), + timeout=timeout, + ) except asyncio.TimeoutError: _kill(process) try: @@ -196,6 +272,7 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, # the NEW process from tracking. if _active_processes.get(agent_key) is process: del _active_processes[agent_key] + _active_started.pop(agent_key, None) async def _run_claude_cli(agent_key: str, prompt: str, timeout: int, role: str, capabilities: str) -> tuple[int, str, str]: @@ -208,7 +285,7 @@ async def _run_claude_cli(agent_key: str, prompt: str, timeout: int, role: str, return await _communicate(agent_key, cmd, prompt.encode("utf-8"), timeout) -async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str, role: str, capabilities: str) -> tuple[int, str, str]: +async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str, role: str, capabilities: str, on_line=None) -> tuple[int, str, str]: cfg = PROVIDERS[provider] # Prompt via temp file instead of argv (ARG_MAX protection for large project prompts) with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8", dir=tempfile.gettempdir()) as f: @@ -224,9 +301,11 @@ async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str "--dangerously-skip-permissions", "-f", str(prompt_path), ] + if on_line is not None: + cmd += ["--format", "json"] # raw JSON events → parsed live by on_line try: - rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True) - return rc, _clean_opencode_output(stdout), stderr + rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True, on_line=on_line) + return rc, (stdout if on_line is not None else _clean_opencode_output(stdout)), stderr finally: prompt_path.unlink(missing_ok=True) diff --git a/backend/blocks.py b/backend/blocks.py index ec83337..3f704d8 100644 --- a/backend/blocks.py +++ b/backend/blocks.py @@ -23,14 +23,14 @@ 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, EMBEDDING_SUB_SAME +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, KANBAN_INVENTORY 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, + _relevance_schema, _semaphore, _str_list, _levels_schema, _timeout, run_single_slot, ) from textkit import ( _unique_title, _load_blocks, _norm_title, _parse_selection, _parse_subblocks, _title, @@ -55,6 +55,13 @@ 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) +RESEARCH_MIN_RUNTIME = 300 # research: do not finish before 5 min (let agents search thoroughly) +RESEARCH_MAX_RUNTIME = 1800 # research: hard wall-clock cap at 30 min + +# Inventory columns (kanban streaming dataflow) — also the fine-step labels shown as pills. +INVENTORY_STEPS = ("Research", "Merge", "Chain", "Chain-Verify", "Naming", "Naming-Verify", + "Chain-Filter", "Filter-Verify", "Block", "Small-Blocks", "Small-Verify", + "Dependency", "Dependency-Verify", "Main-Block") 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 @@ -198,7 +205,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", "Blocks-Filter") + base = INVENTORY_STEPS rest = ( "Subblocks find", "Subblocks select", "Subblocks clarify", "Facts find", "Facts check", "Facts fix", @@ -228,7 +235,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", "Blocks-Filter", "Supplement")), + ("Inventory", INVENTORY_STEPS + ("Supplement",)), ("Subblocks", ("Subblocks find", "Subblocks select", "Subblocks clarify")), ("Facts", ("Facts find", "Facts check", "Facts fix")), ("Levels", ("Levels find", "Levels select", "Levels clarify")), @@ -288,9 +295,11 @@ def _all_slot_files(files: dict) -> list[Path]: # 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 [] + + list(work_dir.glob("research-*")) + + list(work_dir.glob("combine-*")) + list(work_dir.glob("verify-*")) + list(work_dir.glob("naming*")) + # legacy artefacts of the old consolidation/clarification/filter steps (cleaned on reset) + + 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), @@ -317,10 +326,10 @@ 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", "Blocks-Filter"): + for step in ("Source prep",) + INVENTORY_STEPS: 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 + return _step_idx(topic, INVENTORY_STEPS[-1]) # statuses done but artefact gone → rewrite q = load_source(topic) if q["type"] == "projekt" and not files["ergaenzung"].exists(): return _step_idx(topic, "Supplement") @@ -504,27 +513,18 @@ async def _reset_from_step(topic: str, step_idx: int, to_idx: int | None = 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"} & 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 {"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 + # --- Inventory (kanban streaming dataflow) --- + # The kanban flow is a single streaming run (cards, not discrete steps) → any inventory-step reset + # rewinds the WHOLE inventory: clear the kanban tables + the mirrored blocks, research rebuilds. + inv = set(INVENTORY_STEPS) - {"Research"} + if ({"Research"} | inv) & affected: for p_old in _all_slot_files(files): p_old.unlink(missing_ok=True) await db.delete_blocks(topic) + await db.kanban_reset(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). 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: + # status/resume see the inventory as open. On a BOUNDED reset (to_idx set) the later steps are kept. + if to_idx is None and ({"Research"} | inv) & affected: files["final"].unlink(missing_ok=True) @@ -1800,18 +1800,6 @@ def _crawl_index(folder) -> dict[str, str]: -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).""" @@ -2018,7 +2006,8 @@ async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder, "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) + cancelled=is_cancelled, grace=RESEARCH_GRACE, + min_runtime=RESEARCH_MIN_RUNTIME, max_runtime=RESEARCH_MAX_RUNTIME) if is_cancelled(): return False if not agent_texts: @@ -2176,331 +2165,274 @@ def _canonical(candidates: list[dict], idxs: list[int], seen_norm: set[str]) -> return {"title": title, "description": candidates[k]["description"]} -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 - 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 pair_path(pi): return work_dir / f"consolidation-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: {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([_filt(pi, p) for pi, p in enumerate(packages)], - len(packages), _report_p(set_p, topic, "Consolidation")) - if is_cancelled(): +def _naming_schema(data, count: int) -> int | None: + """{"best": N} → 1-based member index in [1, count] · otherwise None.""" + if not isinstance(data, dict): return None - 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] + try: + n = int(data.get("best")) + except (ValueError, TypeError): + return None + return n if 1 <= n <= count else None -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) — pure blocking for recall. - blocks = await asyncio.to_thread(embedding.capped_blocks, sims, None, None) - # 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) - 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 (pairwise): {len(blocks)} blocks ({multi_blocks} multi) " - 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.""" +async def _merge(ctx: GenContext, set_p, files: dict) -> bool: + """Merge (exact dedup): already folded at ingest (upsert by title_norm + reader-union). This + step only restores the candidates after a reset (re-ingest from research-*.md) and marks done.""" topic = ctx.topic - if await db.get_step_status(topic, "Consolidation") == "done": + if await db.get_step_status(topic, "Merge") == "done": 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. + set_p("Merge…", step=_step_idx(topic, "Merge")) + if not await db.list_blocks(topic): await _reingest_research_files(topic, files["arbeit"]) - candidates = await db.list_blocks(topic) - if not candidates: - _blocks_errors[topic] = "Consolidation: no candidates" + if not await db.list_blocks(topic): + _blocks_errors[topic] = "Merge: 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") + await db.set_step_status(topic, "Merge", "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": +async def _combine(ctx: GenContext, set_p, files: dict) -> bool: + """Combine (blocking): embed all candidates, build capped similarity blocks, emit the block- + internal candidate PAIRS (cosine ≥ DEDUP_PAIR_FLOOR) for the Verify judge. Pure embedding, no + LLM. Pairs stored by title_norm. No embedding model → no pairs (every candidate stays its own).""" + topic = ctx.topic + if await db.get_step_status(topic, "Combine") == "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) + set_p("Combine…", step=_step_idx(topic, "Combine")) + work_dir = files["arbeit"] + if not await db.list_blocks(topic): + await _reingest_research_files(topic, work_dir) + rows = await db.list_blocks(topic) + if not rows: + _blocks_errors[topic] = "Combine: no candidates" + return False + by_norm = {r["title_norm"]: r for r in rows} + order = sorted(by_norm) + pairs_norm: list[list[str]] = [] + if EMBEDDING_AKTIV and len(order) >= 2 and await asyncio.to_thread(embedding.available): + texts = [f"{by_norm[nm]['title']} — {by_norm[nm]['description']}" if by_norm[nm]["description"] + else by_norm[nm]["title"] for nm in order] + sims = await asyncio.to_thread(embedding.embed_sims, texts) + if sims is not None: + blocks = await asyncio.to_thread(embedding.capped_blocks, sims, None, None) + 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_norm.append([order[i], order[j]]) + atomic_write_json(work_dir / "combine-pairs.json", {"pairs": pairs_norm}, indent=1) + _log(topic, f"Combine: {len(order)} candidates → {len(pairs_norm)} candidate pairs (embedding blocking)") + await db.set_step_status(topic, "Combine", "done") + return True + + +async def _verify_chains(ctx: GenContext, set_p, files: dict) -> bool: + """Verify (matching): a judge decides each candidate pair (ja/nein), packages run in parallel. + Confirmed pairs become complete-link cliques (no chaining → no aspect over-merge). Output = + chains (groups of title_norms, a partition of all candidates) → verify-chains.json.""" + topic, is_cancelled = ctx.topic, ctx.is_cancelled + if await db.get_step_status(topic, "Verify") == "done": + return True + set_p("Verify…", step=_step_idx(topic, "Verify")) + work_dir = files["arbeit"] + rows = await db.list_blocks(topic) + by_norm = {r["title_norm"]: r for r in rows} + order = sorted(by_norm) + idx_of = {nm: i for i, nm in enumerate(order)} + data = _json_file(work_dir / "combine-pairs.json") + raw_pairs = data.get("pairs", []) if isinstance(data, dict) else [] + pairs = [(a, b) for a, b in raw_pairs if a in idx_of and b in idx_of] # robust against re-ingest + edge_norm: list[tuple[str, str]] = [] + if pairs: + packages = [pairs[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(pairs), DEDUP_PAIRS_CHUNK)] + + def pair_path(pi): return work_dir / f"verify-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"{k + 1}.\nA: {by_norm[a]['title']} — {by_norm[a]['description']}" + f"\nB: {by_norm[b]['title']} — {by_norm[b]['description']}" + for k, (a, b) in enumerate(paare)) + await run_single_slot( + ctx, f"Verify pairs {pi}", + key=f"blocks-{topic}-verify-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, "Verify")) 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)} # 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: - 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") + for pi, paare in enumerate(packages): + verdict = _pairs_schema(_json_file(pair_path(pi))) or {} + for k, (a, b) in enumerate(paare): + if verdict.get(k + 1): + edge_norm.append((a, b)) + edges = [(idx_of[a], idx_of[b]) for a, b in edge_norm] + cliques = _cliques(len(order), edges) + covered = {i for g in cliques for i in g} + groups = cliques + [[i] for i in range(len(order)) if i not in covered] + chains = [[order[i] for i in g] for g in groups] + atomic_write_json(work_dir / "verify-chains.json", {"chains": chains}, indent=1) + multi = sum(1 for c in chains if len(c) > 1) + _log(topic, f"Verify: {len(edge_norm)} confirmed pairs → {len(chains)} chains ({multi} multi)") + await db.set_step_status(topic, "Verify", "done") + return True + + +async def _naming(ctx: GenContext, set_p, files: dict) -> bool: + """Naming (canonicalization): per multi-member chain a judge picks the best, most concrete + EXISTING member title (no invented umbrella term). Singletons keep their title. Output = winner + title_norm per chain → naming.json. Fallback without a model: the _canonical heuristic.""" + topic, is_cancelled = ctx.topic, ctx.is_cancelled + if await db.get_step_status(topic, "Naming") == "done": + return True + set_p("Naming…", step=_step_idx(topic, "Naming")) + work_dir = files["arbeit"] + by_norm = {r["title_norm"]: r for r in await db.list_blocks(topic)} + data = _json_file(work_dir / "verify-chains.json") + chains = [[nm for nm in c if nm in by_norm] for c in (data.get("chains", []) if isinstance(data, dict) else [])] + chains = [c for c in chains if c] + if not chains: + _blocks_errors[topic] = "Naming: no chains" + return False + multi = [(ci, c) for ci, c in enumerate(chains) if len(c) > 1] + + def name_path(ci): return work_dir / f"naming-c{ci}.json" + + async def _name(ci, members): + fp = name_path(ci) + if _naming_schema(_json_file(fp), len(members)) is not None: + return # resume + lines = "\n".join(f"{k + 1}. {by_norm[nm]['title']} — {by_norm[nm]['description']}" + for k, nm in enumerate(members)) + await run_single_slot( + ctx, f"Naming {ci}", + key=f"blocks-{topic}-naming-c{ci}", + prompt=_prompt("Blocks-Naming", topic=topic, members=lines, out_path=fp), + role="judge", capabilities="files", + payload=lambda result, p=fp, n=len(members): _naming_schema(_json_file(p), n), + timeout=_timeout("selection_mapping", len(members)), + ) + + await _gather_progress([_name(ci, c) for ci, c in multi], len(multi), _report_p(set_p, topic, "Naming")) + if is_cancelled(): + return False + result = [] + for ci, members in enumerate(chains): + if len(members) == 1: + winner = members[0] + else: + best = _naming_schema(_json_file(name_path(ci)), len(members)) + if best is not None: + winner = members[best - 1] + else: # judge failed → deterministic _canonical heuristic over the members + rep = _canonical([by_norm[nm] for nm in members], list(range(len(members))), set()) + winner = _norm_title(rep["title"]) + if winner not in members: + winner = members[0] + result.append({"members": members, "winner": winner}) + atomic_write_json(work_dir / "naming.json", {"chains": result}, indent=1) + _log(topic, f"Naming: {len(result)} chains named ({len(multi)} via judge)") + await db.set_step_status(topic, "Naming", "done") + return True + + +async def _verify_naming(ctx: GenContext, set_p, files: dict) -> bool: + """Verify-Naming: a second judge checks each chain's chosen title and corrects it if another + member fits better. Updates naming.json. Singletons skipped.""" + topic, is_cancelled = ctx.topic, ctx.is_cancelled + if await db.get_step_status(topic, "Verify-Naming") == "done": + return True + set_p("Verify-Naming…", step=_step_idx(topic, "Verify-Naming")) + work_dir = files["arbeit"] + by_norm = {r["title_norm"]: r for r in await db.list_blocks(topic)} + data = _json_file(work_dir / "naming.json") + chains = data.get("chains", []) if isinstance(data, dict) else [] + multi = [(ci, [nm for nm in c.get("members", []) if nm in by_norm]) + for ci, c in enumerate(chains) if len([nm for nm in c.get("members", []) if nm in by_norm]) > 1] + if not multi: + await db.set_step_status(topic, "Verify-Naming", "done") + return True + + def check_path(ci): return work_dir / f"naming-check-c{ci}.json" + + async def _check(ci, members, current): + fp = check_path(ci) + if _naming_schema(_json_file(fp), len(members)) is not None: + return # resume + lines = "\n".join(f"{k + 1}. {by_norm[nm]['title']} — {by_norm[nm]['description']}" + for k, nm in enumerate(members)) + await run_single_slot( + ctx, f"Verify-Naming {ci}", + key=f"blocks-{topic}-naming-check-c{ci}", + prompt=_prompt("Blocks-Naming-Check", topic=topic, members=lines, current=current, out_path=fp), + role="judge", capabilities="files", + payload=lambda result, p=fp, n=len(members): _naming_schema(_json_file(p), n), + timeout=_timeout("selection_mapping", len(members)), + ) + + coros = [] + for ci, members in multi: + cur = chains[ci].get("winner") + current = members.index(cur) + 1 if cur in members else 1 + coros.append(_check(ci, members, current)) + await _gather_progress(coros, len(multi), _report_p(set_p, topic, "Verify-Naming")) + if is_cancelled(): + return False + changed = 0 + for ci, members in multi: + best = _naming_schema(_json_file(check_path(ci)), len(members)) + if best is not None and members[best - 1] != chains[ci].get("winner"): + chains[ci]["winner"] = members[best - 1] + changed += 1 + atomic_write_json(work_dir / "naming-final.json", {"chains": chains}, indent=1) + _log(topic, f"Verify-Naming: {changed} titles corrected") + await db.set_step_status(topic, "Verify-Naming", "done") + return True + + +async def _filter(ctx: GenContext, set_p, files: dict) -> bool: + """Filter (reduce): each chain collapses to its winner. Winner → status consensus (the final + block, keeping its own title/description/source), all other members → discarded. Candidates not + covered by any chain stay consensus (no concept loss). Pure Python, no LLM.""" + topic = ctx.topic + if await db.get_step_status(topic, "Filter") == "done": + return True + set_p("Filter…", step=_step_idx(topic, "Filter")) + work_dir = files["arbeit"] + by_norm = {r["title_norm"]: r for r in await db.list_blocks(topic)} + data = _json_file(work_dir / "naming-final.json") or _json_file(work_dir / "naming.json") + chains = data.get("chains", []) if isinstance(data, dict) else [] + if not chains: + _blocks_errors[topic] = "Filter: no chains" + return False + kept, seen = 0, set() + for chain in chains: + members = [nm for nm in chain.get("members", []) if nm in by_norm] + if not members: + continue + winner = chain.get("winner") if chain.get("winner") in members else members[0] + await db.set_block_status(topic, winner, "consensus") + seen.add(winner); kept += 1 + for nm in members: + if nm != winner: + await db.set_block_status(topic, nm, "discarded") + seen.add(nm) + for nm in by_norm: # safety: any uncovered candidate survives + if nm not in seen: + await db.set_block_status(topic, nm, "consensus") + kept += 1 + _log(topic, f"Filter: {kept} final blocks (consensus)") + await db.set_step_status(topic, "Filter", "done") return True @@ -2541,117 +2473,6 @@ def _cliques(n: int, edge_list: list[tuple[int, int]]) -> list[list[int]]: return groups -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"]) @@ -3043,13 +2864,14 @@ 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", "Blocks-Filter"]) + await db.kanban_reset(topic) + await db.delete_pipeline_state(topic, list(INVENTORY_STEPS)) 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: +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, research: bool = True) -> None: if topic in _blocks_progress: return _blocks_progress[topic] = "Waiting…" @@ -3104,31 +2926,27 @@ async def generate_blocks(topic: str, instructions: str = "", provider: str = DE # 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 + # "Continue" (research=False) is non-destructive: never wipe a finished topic. + done = research and 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.kanban_reset(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("Blocks-Filter"): return - if not await _stage(_filter_inventory(ctx, set_p, files)): + # Inventory: streaming kanban dataflow — its columns ARE the inventory steps (pills). + # One streaming run (cards, not discrete steps); per-column regenerate is not meaningful. + import kanban # lazy import: avoids a module-level cycle (kanban imports blocks) + if not await _stage(kanban.run_kanban(ctx, set_p, files, q, folder, instructions, research=research)): return + for _s in INVENTORY_STEPS: # mark all columns done so the step pills show complete + await db.set_step_status(topic, _s, "done") consensus_rows = await db.list_blocks(topic, status="consensus") entries = { i: (f"{b['title']} — {b['description']}" if b["description"] else b["title"]) diff --git a/backend/config.py b/backend/config.py index 992ed39..622f07c 100644 --- a/backend/config.py +++ b/backend/config.py @@ -55,6 +55,9 @@ MAX_CONCURRENT_AGENTS = int(os.getenv("MAX_CONCURRENT_AGENTS", "10")) MAX_CONCURRENT_AGENTS_PER_TOPIC = int(os.getenv("MAX_CONCURRENT_AGENTS_PER_TOPIC", "10")) # per topic MAX_CONCURRENT_INTERACTIVE = 8 +# Inventory engine: streaming kanban dataflow (kanban.py) is the default; set "0" for the legacy ER pipeline. +KANBAN_INVENTORY = os.getenv("KANBAN_INVENTORY", "1") != "0" + # Grace window of the consensus races (blocks, guide, OnePager): after the first # valid result the remaining agents may still become done for this many seconds # (kill only once the minimum is already in). diff --git a/backend/database.py b/backend/database.py index 2ec89c5..9044a12 100644 --- a/backend/database.py +++ b/backend/database.py @@ -199,6 +199,59 @@ CREATE TABLE IF NOT EXISTS sub_artefakte ( ) """ +# Kanban streaming dataflow for the inventory phase. Cards (titles → chains → blocks) flow through +# columns; `stage` is the current/next column (the queue of a worker = WHERE stage = ). +# `stage` is used instead of the reserved word `column`. chain_id/block_id are stable → upsert, not dup. +CREATE_KANBAN_TITLES = """ +CREATE TABLE IF NOT EXISTS kanban_titles ( + topic TEXT NOT NULL, + title_norm TEXT NOT NULL, + title TEXT NOT NULL, + source TEXT NOT NULL DEFAULT '', + content TEXT NOT NULL DEFAULT '', + stage TEXT NOT NULL DEFAULT 'merge', + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, title_norm) +) +""" + +CREATE_KANBAN_CHAINS = """ +CREATE TABLE IF NOT EXISTS kanban_chains ( + topic TEXT NOT NULL, + chain_id TEXT NOT NULL, + stage TEXT NOT NULL DEFAULT 'chain_verify', + main_title_norm TEXT, + dirty INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, chain_id) +) +""" + +CREATE_KANBAN_CHAIN_MEMBERS = """ +CREATE TABLE IF NOT EXISTS kanban_chain_members ( + topic TEXT NOT NULL, + chain_id TEXT NOT NULL, + title_norm TEXT NOT NULL, + PRIMARY KEY (topic, title_norm) +) +""" + +CREATE_KANBAN_BLOCKS = """ +CREATE TABLE IF NOT EXISTS kanban_blocks ( + topic TEXT NOT NULL, + block_id TEXT NOT NULL, + chain_id TEXT, + title TEXT NOT NULL, + source TEXT NOT NULL DEFAULT '', + content TEXT NOT NULL DEFAULT '', + stage TEXT NOT NULL DEFAULT 'small_blocks', + is_small INTEGER NOT NULL DEFAULT 0, + parent_block_id TEXT, + updated_at TEXT NOT NULL, + PRIMARY KEY (topic, block_id) +) +""" + _db: aiosqlite.Connection | None = None @@ -230,6 +283,10 @@ async def init_db(): await db.execute(CREATE_SOURCE) await db.execute(CREATE_GUIDE_OUTLINE) await db.execute(CREATE_SUB_ARTEFAKTE) + await db.execute(CREATE_KANBAN_TITLES) + await db.execute(CREATE_KANBAN_CHAINS) + await db.execute(CREATE_KANBAN_CHAIN_MEMBERS) + await db.execute(CREATE_KANBAN_BLOCKS) try: # migration for existing DBs without the step column await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER") except aiosqlite.OperationalError: @@ -706,6 +763,144 @@ async def delete_blocks(topic: str) -> None: await db.commit() +# ── Kanban streaming dataflow (inventory) ─────────────────────────────────────── +# Generic stage helpers. `stage` is the queue key: a worker pulls WHERE stage = . +_KANBAN_ID = {"kanban_titles": "title_norm", "kanban_chains": "chain_id", "kanban_blocks": "block_id"} + + +async def kanban_pull(topic: str, table: str, stage: str, limit: int) -> list[dict]: + """Oldest `limit` cards sitting in `stage` (FIFO via updated_at).""" + idc = _KANBAN_ID[table] # validates table name + db = await get_db() + cursor = await db.execute( + f"SELECT * FROM {table} WHERE topic = ? AND stage = ? ORDER BY updated_at LIMIT ?", (topic, stage, limit)) + rows = await cursor.fetchall() + return [_row_to_dict(row, cursor) for row in rows] + + +async def kanban_count(topic: str, table: str, stages) -> int: + """How many cards sit in any of `stages` (str or list) — for queue length / quiescence.""" + _ = _KANBAN_ID[table] + if isinstance(stages, str): + stages = [stages] + if not stages: + return 0 + db = await get_db() + ph = ",".join("?" * len(stages)) + cursor = await db.execute(f"SELECT count(*) FROM {table} WHERE topic = ? AND stage IN ({ph})", (topic, *stages)) + return (await cursor.fetchone())[0] + + +async def kanban_advance(topic: str, table: str, id_val: str, stage: str) -> None: + """Move a card to `stage` (advance to next column, or back for rework/retraction).""" + idc = _KANBAN_ID[table] + db = await get_db() + await db.execute(f"UPDATE {table} SET stage = ?, updated_at = ? WHERE topic = ? AND {idc} = ?", + (stage, _now(), topic, id_val)) + await db.commit() + + +async def kanban_add_title(topic: str, title_norm: str, title: str, source: str = "", content: str = "") -> bool: + """Research → titles queue (stage 'merge'). Exact dupes are dropped (PK conflict). → True if new.""" + db = await get_db() + cursor = await db.execute( + """INSERT INTO kanban_titles (topic, title_norm, title, source, content, stage, updated_at) + VALUES (?, ?, ?, ?, ?, 'merge', ?) ON CONFLICT(topic, title_norm) DO NOTHING""", + (topic, title_norm, title, source, content, _now())) + await db.commit() + return cursor.rowcount > 0 + + +async def kanban_upsert_chain(topic: str, chain_id: str, stage: str, main_title_norm: str | None = None, + dirty: int = 0) -> None: + db = await get_db() + await db.execute( + """INSERT INTO kanban_chains (topic, chain_id, stage, main_title_norm, dirty, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(topic, chain_id) DO UPDATE SET + stage = excluded.stage, main_title_norm = COALESCE(excluded.main_title_norm, kanban_chains.main_title_norm), + dirty = excluded.dirty, updated_at = excluded.updated_at""", + (topic, chain_id, stage, main_title_norm, dirty, _now())) + await db.commit() + + +async def kanban_set_chain_members(topic: str, chain_id: str, members: list[str]) -> None: + """Replace the member set of a chain (one title belongs to exactly one chain).""" + db = await get_db() + await db.execute("DELETE FROM kanban_chain_members WHERE topic = ? AND chain_id = ?", (topic, chain_id)) + for nm in members: + await db.execute( + """INSERT INTO kanban_chain_members (topic, chain_id, title_norm) VALUES (?, ?, ?) + ON CONFLICT(topic, title_norm) DO UPDATE SET chain_id = excluded.chain_id""", + (topic, chain_id, nm)) + await db.commit() + + +async def kanban_chain_members(topic: str, chain_id: str) -> list[str]: + db = await get_db() + cursor = await db.execute( + "SELECT title_norm FROM kanban_chain_members WHERE topic = ? AND chain_id = ?", (topic, chain_id)) + return [r[0] for r in await cursor.fetchall()] + + +async def kanban_member_chain(topic: str, title_norm: str) -> str | None: + """Which chain a title currently belongs to (or None).""" + db = await get_db() + cursor = await db.execute( + "SELECT chain_id FROM kanban_chain_members WHERE topic = ? AND title_norm = ?", (topic, title_norm)) + row = await cursor.fetchone() + return row[0] if row else None + + +async def kanban_upsert_block(topic: str, block_id: str, chain_id: str | None, title: str, source: str = "", + content: str = "", stage: str = "small_blocks", is_small: int = 0, + parent_block_id: str | None = None) -> None: + """Chain-id-stable block (Filter/Block). Upsert → growing chains overwrite, never duplicate.""" + db = await get_db() + await db.execute( + """INSERT INTO kanban_blocks (topic, block_id, chain_id, title, source, content, stage, is_small, parent_block_id, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(topic, block_id) DO UPDATE SET + chain_id = excluded.chain_id, title = excluded.title, source = excluded.source, + content = excluded.content, stage = excluded.stage, is_small = excluded.is_small, + parent_block_id = excluded.parent_block_id, updated_at = excluded.updated_at""", + (topic, block_id, chain_id, title, source, content, stage, is_small, parent_block_id, _now())) + await db.commit() + + +async def kanban_titles_by_norm(topic: str) -> dict[str, dict]: + """All titles of a topic keyed by title_norm (the candidate universe for chaining).""" + db = await get_db() + cursor = await db.execute("SELECT * FROM kanban_titles WHERE topic = ?", (topic,)) + rows = await cursor.fetchall() + return {(d := _row_to_dict(row, cursor))["title_norm"]: d for row in rows} + + +async def kanban_all_blocks(topic: str) -> list[dict]: + db = await get_db() + cursor = await db.execute("SELECT * FROM kanban_blocks WHERE topic = ?", (topic,)) + rows = await cursor.fetchall() + return [_row_to_dict(row, cursor) for row in rows] + + +async def kanban_stage_counts(topic: str) -> dict[str, int]: + """{stage: count} across all kanban tables — for the live board / quiescence.""" + db = await get_db() + out: dict[str, int] = {} + for table in ("kanban_titles", "kanban_chains", "kanban_blocks"): + cursor = await db.execute(f"SELECT stage, count(*) FROM {table} WHERE topic = ? GROUP BY stage", (topic,)) + for stage, n in await cursor.fetchall(): + out[stage] = out.get(stage, 0) + n + return out + + +async def kanban_reset(topic: str) -> None: + db = await get_db() + for table in ("kanban_titles", "kanban_chains", "kanban_chain_members", "kanban_blocks"): + await db.execute(f"DELETE FROM {table} WHERE topic = ?", (topic,)) + await db.commit() + + async def upsert_subblock(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str) -> None: db = await get_db() await db.execute( diff --git a/backend/kanban.py b/backend/kanban.py new file mode 100644 index 0000000..49fc69d --- /dev/null +++ b/backend/kanban.py @@ -0,0 +1,748 @@ +"""Streaming kanban dataflow for the inventory phase. + +Each column is a worker that pulls cards from its input `stage` (the queue), processes up to +KANBAN_BATCH at a time, and advances them to the next stage. Cards: titles → chains → blocks. + +Streaming columns run continuously; barrier columns start only at QUIESCENCE of everything before +them (no active worker + empty queues). Verify columns push failures back (rework). The Chain column +re-clusters live: a chain that gains a member is marked dirty and flows back to chain_verify. + +Reused from blocks.py (imported lazily-safe — kanban is only imported after blocks is loaded): +embedding clustering, `_pairs_schema`/`_cliques`, `_canonical`, research prompt + file payload. +""" + +import asyncio +import json +import uuid + +import database as db +import embedding +import blocks +from config import RESEARCH_GRACE, MAX_CONCURRENT_AGENTS_PER_TOPIC +from pipeline import GenContext, run_single_slot, _prompt, _timeout, _log, OK +from textkit import _norm_title, _title, _parse_selection +from jsonio import read_json_file as _json_file + +KANBAN_BATCH = 5 # cards a worker pulls per package (micro-batching) +# How many packages ONE worker keeps in flight at once. A worker no longer blocks on a single +# package — it keeps pulling and dispatching until this many run concurrently, so a busy column +# fills the agent slots (the per-topic semaphore is the real cap; over-dispatch just queues cheaply). +WORKER_INFLIGHT = MAX_CONCURRENT_AGENTS_PER_TOPIC +# Stages whose processor mutates shared cross-card state and MUST run one package at a time. +# Online chain-clustering reads the whole universe + membership; parallel packages would race. +_SERIAL_STAGES = {"chain"} +_ID_COL = {"kanban_titles": "title_norm", "kanban_chains": "chain_id", "kanban_blocks": "block_id"} +CHAIN_CAP = 12 # max members per chain — caps the O(n²) pair-verification blow-up +_POLL = 0.3 # seconds between empty-queue polls + +# Stage order. A card's `stage` = the column it waits in (its worker's input). +TITLE_STAGES = ["merge", "chain", "chained"] # 'chained' = consumed into a chain +CHAIN_STAGES = ["chain_verify", "naming", "naming_verify", "chain_filter", "filter_verify", "block_assemble"] +BLOCK_STAGES = ["small_blocks", "small_verify", "dependency", "dependency_verify", "main"] +DONE_CHAIN = "done_chain" +DONE_BLOCK = "done_block" +REJECTED = "rejected" # block dropped by filter_verify (off-topic / noise) — terminal, never mirrored + +# Predecessor stages for each barrier (must ALL be quiescent before the barrier worker runs). +_BEFORE_CHAIN_FILTER = ["merge", "chain", "chain_verify", "naming", "naming_verify"] +_BEFORE_BLOCK = _BEFORE_CHAIN_FILTER + ["chain_filter", "filter_verify"] +_BEFORE_MAIN = ["small_blocks", "small_verify", "dependency", "dependency_verify"] + + +class _Flow: + """Shared runtime state: active-task counters per stage + a wakeup event. `producers` counts the + running research agents (initial + any added live via the generate button); research counts as + done only when ALL producers have finished, so the flow stays awake while extras still search.""" + def __init__(self, topic: str, work_dir): + self.topic = topic + self.work_dir = work_dir + self.active: dict[str, int] = {} + self.producers = 1 # the initial research agent + self.research_tag = 0 + self.stop = False + self.wake = asyncio.Event() + self.spawn_research = None # set by run_kanban: () → coroutine that adds one more research agent + + @property + def research_done(self) -> bool: + return self.producers <= 0 + + def add_producer(self): + self.producers += 1 + self.wake.set() + + def done_producer(self): + self.producers -= 1 + self.wake.set() + + def next_tag(self) -> int: + self.research_tag += 1 + return self.research_tag + + def enter(self, stage: str): + self.active[stage] = self.active.get(stage, 0) + 1 + + def leave(self, stage: str): + self.active[stage] = max(0, self.active.get(stage, 0) - 1) + self.wake.set() + + def active_in(self, stages) -> bool: + return any(self.active.get(s, 0) > 0 for s in stages) + + async def queued_in(self, table: str, stages) -> bool: + return await db.kanban_count(self.topic, table, list(stages)) > 0 + + +# ── Research producer ──────────────────────────────────────────────────────────── +async def _ingest_titles(topic: str, text: str) -> int: + """Parse a reader file into kanban_titles (stage 'merge'). Exact dupes drop on the PK. → new count.""" + n, seen = 0, set() + for record in _parse_selection(text).values(): + title = _title(record) + norm = _norm_title(title) + if not norm or norm in seen: + continue + seen.add(norm) + parts = [t.strip() for t in record.split(" — ")] + source = parts[2] if len(parts) >= 3 else "" + desc = parts[1] if len(parts) >= 2 else "" + if await db.kanban_add_title(topic, norm, title, source, desc): + n += 1 + return n + + +RESEARCH_RUNTIME = 900 # one research agent, one round, ~15 min hard cap — the tail ingests live while it writes +_POLL_RESEARCH = 3 # seconds between live reads of a running research file + +# Live registry of running flows, so the "+ research" button can attach another agent to a live run. +_active_flows: dict[str, "_Flow"] = {} + + +def _extract_text(raw_line: str) -> str: + """Best-effort: pull assistant/tool text out of ONE opencode `--format json` event line. + Recursively collects every `text`/`content` string — robust to the exact event schema.""" + try: + obj = json.loads(raw_line) + except Exception: + return "" + parts: list[str] = [] + def _walk(o): + if isinstance(o, dict): + for k, v in o.items(): + if k in ("text", "content") and isinstance(v, str): + parts.append(v) + else: + _walk(v) + elif isinstance(o, list): + for v in o: + _walk(v) + _walk(obj) + return "".join(parts) + + +async def _research_once(ctx: GenContext, files: dict, q: dict, folder, instructions: str, tag: str, flow: "_Flow"): + """ONE agent searches the topic; its titles go into the merge queue LIVE. Two sources feed the + ingest: the JSON event stream (on_line → text buffer) AND the file the agent writes — whichever + the agent uses, cards stream in immediately (not only after it finishes).""" + work_dir = files["arbeit"] + caps = "files" if folder else "full" + p = work_dir / f"research-{tag}.md" + p.unlink(missing_ok=True) + stop = asyncio.Event() + buf: list[str] = [] # assistant text streamed live from the JSON events + + def _on_line(raw: str): # sync, called per stdout line by the agent runner + if (t := _extract_text(raw)): + buf.append(t) + + async def _drain() -> bool: # ingest from BOTH event buffer and file (idempotent, dupes drop on PK) + text = "".join(buf) + if (ft := blocks._file_payload(p)): + text += "\n" + ft + return bool(text) and await _ingest_titles(ctx.topic, text) + + async def _tail(): # live-ingest loop while the agent runs + while not stop.is_set(): + try: + await asyncio.wait_for(stop.wait(), timeout=_POLL_RESEARCH) + except asyncio.TimeoutError: + pass + if await _drain(): + flow.wake.set() # new cards → wake the workers + + tail = asyncio.create_task(_tail()) + try: + await run_single_slot( + ctx, f"research-{tag}", key=f"blocks-{ctx.topic}-research-{tag}", + prompt=blocks._build_research_prompt(ctx.topic, p, instructions, q["type"], folder), + role="quick", capabilities=caps, + payload=(lambda result, p=p: blocks._file_payload(p)), + timeout=RESEARCH_RUNTIME, on_line=_on_line, + ) + finally: + stop.set() + await tail + if await _drain(): # final catch-up + flow.wake.set() + _log(ctx.topic, f"Research {tag}: titles → merge queue") + + +async def _research(ctx: GenContext, files: dict, q: dict, folder, instructions: str, flow: _Flow): + """The initial research producer (already counted in flow.producers=1).""" + try: + await _research_once(ctx, files, q, folder, instructions, "1", flow) + finally: + flow.done_producer() + + +async def _extra_research(ctx: GenContext, files: dict, q: dict, folder, instructions: str, flow: _Flow): + """One more research agent, added live via the generate button. Keeps the flow awake until done.""" + flow.add_producer() + try: + await _research_once(ctx, files, q, folder, instructions, f"x{flow.next_tag()}", flow) + finally: + flow.done_producer() + + +def add_research_agent(topic: str) -> bool: + """Attach one more research agent to a running flow. → True if a run was live to attach to.""" + flow = _active_flows.get(topic) + if flow is None or flow.stop or flow.spawn_research is None: + return False + asyncio.create_task(flow.spawn_research()) + return True + + +# ── Generic worker loop ──────────────────────────────────────────────────────────── +async def _quiescent(flow: _Flow, stages) -> bool: + """True iff no worker is active in `stages` AND no card is queued in any of them (all tables). + The barrier/exit condition — must include QUEUED cards, not just active workers, or a worker + could exit in a momentary lull while an upstream worker still has work to push down.""" + if not stages: + return True + if flow.active_in(stages): + return False + for tb in ("kanban_titles", "kanban_chains", "kanban_blocks"): + if await db.kanban_count(flow.topic, tb, list(stages)): + return False + return True + + +async def _worker(flow: _Flow, table: str, in_stage: str, process, upstream, *, barrier=False, inflight=WORKER_INFLIGHT): + """Pull cards from `in_stage`, run `process` — keeping up to `inflight` packages running CONCURRENTLY + so a busy column fills the agent slots instead of doing one package at a time. `upstream` = all stages + before this one. A barrier worker only pulls when `upstream` is fully quiescent. ANY worker exits only + when research is done, its own queue is empty, AND `upstream` is quiescent (nothing can still arrive). + + Double-pull safety: each stage has exactly ONE worker, so an in-memory `claimed` set of card-ids (held + while a package runs) is enough to keep concurrent pulls from grabbing the same cards.""" + topic = flow.topic + idc = _ID_COL[table] + claimed: set[str] = set() + tasks: set[asyncio.Task] = set() + + async def _run(cards): + ids = [c[idc] for c in cards] + flow.enter(in_stage) + try: + await process(cards) + except Exception as e: # one bad package must not kill the worker + _log(topic, f"worker {in_stage}: {type(e).__name__}: {e}") + finally: + flow.leave(in_stage) + for i in ids: + claimed.discard(i) + flow.wake.set() + + try: + while not flow.stop: + tasks = {t for t in tasks if not t.done()} + # Fill the pipeline: pull fresh cards and dispatch until `inflight` packages run. + if not barrier or await _quiescent(flow, upstream): + while len(tasks) < inflight: + rows = await db.kanban_pull(topic, table, in_stage, KANBAN_BATCH + len(claimed)) + fresh = [r for r in rows if r[idc] not in claimed][:KANBAN_BATCH] + if not fresh: + break + for r in fresh: + claimed.add(r[idc]) + tasks.add(asyncio.create_task(_run(list(fresh)))) + if tasks: # busy → wait for a package to finish, then refill + await asyncio.wait(tasks, timeout=_POLL, return_when=asyncio.FIRST_COMPLETED) + continue + # idle: nothing in flight and nothing pulled + up_quiet = await _quiescent(flow, upstream) + if (flow.research_done and up_quiet and not flow.active_in([in_stage]) + and await db.kanban_count(topic, table, in_stage) == 0): + return # nothing left and nothing upstream can produce + await _sleep_wake(flow) + finally: + for t in tasks: + t.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + +async def _sleep_wake(flow: _Flow): + try: + await asyncio.wait_for(flow.wake.wait(), timeout=_POLL) + except asyncio.TimeoutError: + pass + flow.wake.clear() + + +# ── Column processors ────────────────────────────────────────────────────────────── +async def _proc_merge(flow: _Flow, cards): + """Exact dedup happened at ingest (PK). Merge just advances titles to the chain column.""" + for c in cards: + await db.kanban_advance(flow.topic, "kanban_titles", c["title_norm"], "chain") + flow.wake.set() + + +async def _proc_chain(flow: _Flow, cards): + """Embedding blocking: for each new title, find the most similar existing title (cosine ≥ floor). + Join its chain (or open a new one), mark the chain dirty → chain_verify. Live-growing clusters.""" + topic = flow.topic + by_norm = await db.kanban_titles_by_norm(topic) + # Universe = titles already chained + the new batch (for nearest-neighbour search). + universe = [nm for nm, r in by_norm.items() if r["stage"] in ("chain", "chained")] + if len(universe) < 1: + return + texts = [f"{by_norm[nm]['title']} — {by_norm[nm]['content']}" if by_norm[nm]["content"] else by_norm[nm]["title"] + for nm in universe] + sims = await asyncio.to_thread(embedding.embed_sims, texts) if ( + embedding and await asyncio.to_thread(embedding.available)) else None + idx = {nm: i for i, nm in enumerate(universe)} + # existing membership + member_chain = {} + for nm in universe: + cid = await _chain_of(topic, nm) + if cid: + member_chain[nm] = cid + touched = set() + for c in cards: + nm = c["title_norm"] + target = None + if sims is not None and nm in idx: + best, bestcos = None, blocks.DEDUP_PAIR_FLOOR + for other in universe: + if other == nm or other not in member_chain and other not in idx: + continue + cos = float(sims[idx[nm]][idx[other]]) if other in idx else -1 + if cos >= bestcos and other != nm: + best, bestcos = other, cos + if best is not None: + target = member_chain.get(best) + cid = target or f"c-{uuid.uuid4().hex[:12]}" + members = set(await db.kanban_chain_members(topic, cid)) + if target and len(members) >= CHAIN_CAP: # neighbour's chain is full → start a fresh chain + cid = f"c-{uuid.uuid4().hex[:12]}" + members = set() + members.add(nm) + await db.kanban_set_chain_members(topic, cid, sorted(members)) + await db.kanban_upsert_chain(topic, cid, "chain_verify", dirty=1) + member_chain[nm] = cid + await db.kanban_advance(topic, "kanban_titles", nm, "chained") + touched.add(cid) + flow.wake.set() + + +async def _chain_of(topic: str, title_norm: str) -> str | None: + return await db.kanban_member_chain(topic, title_norm) + + +async def _members_dicts(topic: str, members: list[str]) -> list[dict]: + by = await db.kanban_titles_by_norm(topic) + return [by[m] for m in members if m in by] + + +async def _proc_chain_verify(ctx: GenContext, flow: _Flow, cards): + """Pairwise-verify the batch's chains IN PARALLEL (one agent per chain). Failures split off.""" + await asyncio.gather(*[_verify_one(ctx, flow, c) for c in cards], return_exceptions=True) + flow.wake.set() + + +async def _verify_one(ctx: GenContext, flow: _Flow, c): + topic = flow.topic + cid = c["chain_id"] + members = await db.kanban_chain_members(topic, cid) + dicts = await _members_dicts(topic, members) + if len(dicts) <= 1: + await db.kanban_upsert_chain(topic, cid, "naming", dirty=0) + return + # Only the embedding-NEAR candidate pairs (cosine ≥ floor) — NOT all O(n²) pairs. A 12-member + # chain shrinks from 66 pairs to a handful. Transitivity (connected components) does the rest. + nm = [d["title_norm"] for d in dicts] + texts = [f"{d['title']} — {d['content']}" if d["content"] else d["title"] for d in dicts] + sims = await asyncio.to_thread(embedding.embed_sims, texts) if ( + embedding and await asyncio.to_thread(embedding.available)) else None + if sims is not None: + pairs = [(nm[i], nm[j]) for i in range(len(nm)) for j in range(i + 1, len(nm)) + if float(sims[i][j]) >= blocks.DEDUP_PAIR_FLOOR] + else: + pairs = [(a, b) for x, a in enumerate(members) for b in members[x + 1:]] + keep_edges = await _verify_pairs(ctx, flow.work_dir, topic, cid, dicts, pairs) + groups = _components(members, keep_edges) # transitive groups over confirmed near-pairs + groups.sort(key=len, reverse=True) + main = groups[0] if groups else members + await db.kanban_set_chain_members(topic, cid, sorted(main)) + await db.kanban_upsert_chain(topic, cid, "naming", dirty=0) + for g in groups[1:]: # the rest split into fresh chains + ncid = f"c-{uuid.uuid4().hex[:12]}" + await db.kanban_set_chain_members(topic, ncid, sorted(g)) + await db.kanban_upsert_chain(topic, ncid, "naming", dirty=0) + + +def _components(members: list[str], edges) -> list[list[str]]: + """Connected components (union-find) over confirmed pairs. Members without an edge stay alone.""" + parent = {m: m for m in members} + + def find(x): + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + for a, b in edges: + if a in parent and b in parent: + parent[find(a)] = find(b) + comp: dict[str, list[str]] = {} + for m in members: + comp.setdefault(find(m), []).append(m) + return list(comp.values()) + + +async def _verify_pairs(ctx, work_dir, topic, cid, dicts, pairs): + """Judge the candidate pairs in DEDUP_PAIRS_CHUNK packages, all packages IN PARALLEL → confirmed edges.""" + by = {d["title_norm"]: d for d in dicts} + chunks = [pairs[k:k + blocks.DEDUP_PAIRS_CHUNK] for k in range(0, len(pairs), blocks.DEDUP_PAIRS_CHUNK)] + + async def _chunk(ci, chunk): + path = work_dir / f"verify-{cid}-{ci}.json" + lines = "\n\n".join( + f"{j + 1}.\nA: {by[a]['title']} — {by[a]['content']}\nB: {by[b]['title']} — {by[b]['content']}" + for j, (a, b) in enumerate(chunk)) + await run_single_slot( + ctx, f"Chain verify {cid}", key=f"blocks-{topic}-verify-{cid}-{ci}", + prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=path), + role="judge", capabilities="files", + payload=lambda result, p=path: blocks._pairs_schema(_json_file(p)), + timeout=_timeout("selection_mapping", len(chunk))) + verdict = blocks._pairs_schema(_json_file(path)) or {} + return [(a, b) for j, (a, b) in enumerate(chunk) if verdict.get(j + 1)] + + results = await asyncio.gather(*[_chunk(ci, ch) for ci, ch in enumerate(chunks)], return_exceptions=True) + return [e for r in results if isinstance(r, list) for e in r] + + +async def _proc_naming(ctx: GenContext, flow: _Flow, cards): + """Pick the best member title per chain — batch runs IN PARALLEL (one agent per chain).""" + await asyncio.gather(*[_name_one(ctx, flow, c) for c in cards], return_exceptions=True) + flow.wake.set() + + +async def _name_one(ctx: GenContext, flow: _Flow, c): + topic = flow.topic + cid = c["chain_id"] + members = await db.kanban_chain_members(topic, cid) + dicts = await _members_dicts(topic, members) + if len(dicts) <= 1: + await db.kanban_upsert_chain(topic, cid, "naming_verify", + main_title_norm=(members[0] if members else None), dirty=0) + return + winner = await _choose_title(ctx, flow.work_dir, topic, cid, members, dicts, "Blocks-Naming") + await db.kanban_upsert_chain(topic, cid, "naming_verify", main_title_norm=winner, dirty=0) + + +async def _proc_naming_verify(ctx: GenContext, flow: _Flow, cards): + """Second judge checks each title — batch runs IN PARALLEL.""" + await asyncio.gather(*[_namecheck_one(ctx, flow, c) for c in cards], return_exceptions=True) + flow.wake.set() + + +async def _namecheck_one(ctx: GenContext, flow: _Flow, c): + topic = flow.topic + cid = c["chain_id"] + members = await db.kanban_chain_members(topic, cid) + dicts = await _members_dicts(topic, members) + winner = c.get("main_title_norm") or (members[0] if members else None) + if len(dicts) > 1: + winner = await _choose_title(ctx, flow.work_dir, topic, cid, members, dicts, "Blocks-Naming-Check", + current=(members.index(winner) + 1 if winner in members else 1)) + await db.kanban_upsert_chain(topic, cid, "chain_filter", main_title_norm=winner, dirty=0) + + +async def _choose_title(ctx, work_dir, topic, cid, members, dicts, template, current=None): + by = {d["title_norm"]: d for d in dicts} + path = work_dir / f"naming-{cid}.json" + lines = "\n".join(f"{k + 1}. {by[m]['title']} — {by[m]['content']}" for k, m in enumerate(members) if m in by) + kw = dict(topic=topic, members=lines, out_path=path) + if current is not None: + kw["current"] = current + await run_single_slot( + ctx, f"Naming {cid}", key=f"blocks-{topic}-naming-{cid}", + prompt=_prompt(template, **kw), role="judge", capabilities="files", + payload=lambda result, p=path: blocks._naming_schema(_json_file(p), len(members)), + timeout=_timeout("selection_mapping", len(members))) + best = blocks._naming_schema(_json_file(path), len(members)) + if best is None: + rep = blocks._canonical(dicts, list(range(len(dicts))), set()) + w = _norm_title(rep["title"]) + return w if w in members else members[0] + return members[best - 1] + + +async def _proc_chain_filter(flow: _Flow, cards): + """BARRIER. Reduce each chain to its winner → upsert a block (chain_id stable). → filter_verify.""" + topic = flow.topic + by = await db.kanban_titles_by_norm(topic) + for c in cards: + cid = c["chain_id"] + members = await db.kanban_chain_members(topic, cid) + winner = c.get("main_title_norm") if c.get("main_title_norm") in members else (members[0] if members else None) + if not winner or winner not in by: + await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0) + continue + w = by[winner] + await db.kanban_upsert_block(topic, f"b-{cid}", cid, w["title"], w["source"], w["content"], stage="filter_verify_b") + await db.kanban_upsert_chain(topic, cid, "filter_verify", dirty=0) + flow.wake.set() + + +async def _proc_filter_verify(ctx: GenContext, flow: _Flow, cards): + """An agent confirms each reduced block is a valid, on-topic, self-contained concept. Off-topic / + noise / empty blocks are dropped (→ REJECTED, chain done). IN PARALLEL (one agent per block).""" + await asyncio.gather(*[_filtercheck_one(ctx, flow, c) for c in cards], return_exceptions=True) + flow.wake.set() + + +async def _filtercheck_one(ctx: GenContext, flow: _Flow, c): + topic = flow.topic + cid = c["chain_id"] + bid = f"b-{cid}" + by = await db.kanban_titles_by_norm(topic) + members = await db.kanban_chain_members(topic, cid) + winner = c.get("main_title_norm") if c.get("main_title_norm") in by else (members[0] if members else None) + + async def _drop(): + await db.kanban_advance(topic, "kanban_blocks", bid, REJECTED) + await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0) + + async def _pass(): + await db.kanban_advance(topic, "kanban_blocks", bid, "block_assemble_b") + await db.kanban_upsert_chain(topic, cid, "block_assemble", dirty=0) + + if not winner or winner not in by: # nothing to verify → drop the empty chain + await _drop() + return + w = by[winner] + path = flow.work_dir / f"filtercheck-{cid}.json" + await run_single_slot( + ctx, f"Filter verify {cid}", key=f"blocks-{topic}-verify-filter-{cid}", + prompt=_prompt("Blocks-Filter-Check", topic=topic, title=w["title"], content=w["content"], out_path=path), + role="judge", capabilities="files", + payload=lambda result, p=path: _keep_schema(_json_file(p)), + timeout=_timeout("selection_mapping", 1)) + keep = _keep_schema(_json_file(path)) + await (_drop() if keep is False else _pass()) # None (parse fail) → keep, conservative + + +async def _proc_block(flow: _Flow, cards): + """BARRIER. Assemble the final block row → small_blocks. (chain card consumed → done.)""" + topic = flow.topic + for c in cards: + cid = c["chain_id"] + await db.kanban_advance(topic, "kanban_blocks", f"b-{cid}", "small_blocks") + await db.kanban_upsert_chain(topic, cid, DONE_CHAIN, dirty=0) + flow.wake.set() + + +def _small_schema(data, count): + """{"small": {"1": true, ...}} → {block_index: bool} · else None.""" + if not isinstance(data, dict) or not isinstance(data.get("small"), dict): + return None + out = {} + for k, v in data["small"].items(): + try: + n = int(k) + except (ValueError, TypeError): + continue + if 1 <= n <= count: + out[n] = str(v).strip().casefold() in ("true", "ja", "yes", "1") + return out or None + + +def _keep_schema(data): + """{"keep": true/false} → bool · None when absent/unparseable (caller keeps on None, conservative).""" + if not isinstance(data, dict) or "keep" not in data: + return None + return str(data["keep"]).strip().casefold() in ("true", "ja", "yes", "1") + + +def _dep_schema(data, count): + """{"parent": N} → 0..count (0 = standalone) · else None.""" + if not isinstance(data, dict): + return None + try: + n = int(data.get("parent")) + except (ValueError, TypeError): + return None + return n if 0 <= n <= count else None + + +async def _proc_small(ctx: GenContext, flow: _Flow, cards): + """Judge marks fragment-like blocks (batch). → small_verify.""" + topic = flow.topic + path = flow.work_dir / f"small-{cards[0]['block_id']}.json" + lines = "\n".join(f"{i + 1}. {c['title']} — {c['content']}" for i, c in enumerate(cards)) + await run_single_slot( + ctx, "Small blocks", key=f"blocks-{topic}-small-{cards[0]['block_id']}", + prompt=_prompt("Blocks-Small", topic=topic, blocks=lines, out_path=path), + role="judge", capabilities="files", + payload=lambda result, p=path: _small_schema(_json_file(p), len(cards)), + timeout=_timeout("selection_mapping", len(cards))) + verdict = _small_schema(_json_file(path), len(cards)) or {} + for i, c in enumerate(cards): + is_small = 1 if verdict.get(i + 1) else 0 + await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"], + stage="small_verify", is_small=is_small, parent_block_id=c.get("parent_block_id")) + flow.wake.set() + + +async def _proc_small_verify(ctx: GenContext, flow: _Flow, cards): + """Second judge re-checks the small flag (consensus): a block stays `small` only if it was marked + small AND this judge also calls it a fragment. Disagreement → keep as a main block (conservative). + → dependency.""" + topic = flow.topic + path = flow.work_dir / f"smallcheck-{cards[0]['block_id']}.json" + lines = "\n".join(f"{i + 1}. {c['title']} — {c['content']}" for i, c in enumerate(cards)) + await run_single_slot( + ctx, "Small verify", key=f"blocks-{topic}-verify-small-{cards[0]['block_id']}", + prompt=_prompt("Blocks-Small", topic=topic, blocks=lines, out_path=path), + role="judge", capabilities="files", + payload=lambda result, p=path: _small_schema(_json_file(p), len(cards)), + timeout=_timeout("selection_mapping", len(cards))) + verdict = _small_schema(_json_file(path), len(cards)) or {} + for i, c in enumerate(cards): + is_small = 1 if (c["is_small"] and verdict.get(i + 1)) else 0 + await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"], + stage="dependency", is_small=is_small, parent_block_id=c.get("parent_block_id")) + flow.wake.set() + + +async def _proc_dependency(ctx: GenContext, flow: _Flow, cards): + """For each SMALL block, a judge picks its parent from the full list — batch runs IN PARALLEL.""" + topic = flow.topic + parents = [b for b in await db.kanban_all_blocks(topic) if not b["is_small"] and b["stage"] != REJECTED] + plist = "\n".join(f"{i + 1}. {b['title']}" for i, b in enumerate(parents)) + await asyncio.gather(*[_dep_one(ctx, flow, c, parents, plist) for c in cards], return_exceptions=True) + flow.wake.set() + + +async def _dep_one(ctx: GenContext, flow: _Flow, c, parents, plist): + topic = flow.topic + if not c["is_small"] or not parents: + await db.kanban_advance(topic, "kanban_blocks", c["block_id"], "dependency_verify") + return + path = flow.work_dir / f"dep-{c['block_id']}.json" + await run_single_slot( + ctx, "Dependency", key=f"blocks-{topic}-dep-{c['block_id']}", + prompt=_prompt("Blocks-Dependency", topic=topic, + small=f"{c['title']} — {c['content']}", parents=plist, out_path=path), + role="judge", capabilities="files", + payload=lambda result, p=path: _dep_schema(_json_file(p), len(parents)), + timeout=_timeout("selection_mapping", len(parents))) + pick = _dep_schema(_json_file(path), len(parents)) + parent_id = parents[pick - 1]["block_id"] if pick else None + await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"], + stage="dependency_verify", is_small=c["is_small"], parent_block_id=parent_id) + + +async def _proc_dependency_verify(flow: _Flow, cards): + """A small block without a parent is demarked → becomes a main block. → main.""" + topic = flow.topic + for c in cards: + is_small = c["is_small"] + if is_small and not c.get("parent_block_id"): + is_small = 0 + await db.kanban_upsert_block(topic, c["block_id"], c["chain_id"], c["title"], c["source"], c["content"], + stage="main", is_small=is_small, parent_block_id=c.get("parent_block_id")) + flow.wake.set() + + +async def _proc_main(flow: _Flow, cards): + """BARRIER. Finalize non-small blocks → mirror into the legacy `blocks` table as consensus.""" + topic = flow.topic + for c in cards: + if not c["is_small"]: + norm = _norm_title(c["title"]) + await db.upsert_block(topic, norm, c["title"], c["content"], [c["source"]] if c["source"] else []) + await db.set_block_status(topic, norm, "consensus") + await db.kanban_advance(topic, "kanban_blocks", c["block_id"], DONE_BLOCK) + flow.wake.set() + + +# ── Orchestration ────────────────────────────────────────────────────────────────── +async def run_kanban(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str, + research: bool = True) -> bool: + """Run the streaming inventory. Returns True when the whole flow reaches quiescence at 'main'. + + research=False ("Continue"): process the EXISTING queue without searching new titles. No initial + research producer, producers=0 → research_done is true at once; workers drain the queue and exit. + The +Research button can still attach an agent later via flow.spawn_research.""" + topic = ctx.topic + flow = _Flow(topic, files["arbeit"]) + flow.spawn_research = lambda: _extra_research(ctx, files, q, folder, instructions, flow) + if not research: + flow.producers = 0 # continue the existing queue, search no new titles + _active_flows[topic] = flow + set_p("Kanban inventory…") + + ORDER = ["merge", "chain", "chain_verify", "naming", "naming_verify", "chain_filter", + "filter_verify", "block_assemble", "small_blocks", "small_verify", + "dependency", "dependency_verify", "main"] + up = {s: ORDER[:i] for i, s in enumerate(ORDER)} # upstream = all stages before this one + barriers = {"chain_filter", "block_assemble", "main"} + specs = [ + ("kanban_titles", "merge", lambda cs: _proc_merge(flow, cs)), + ("kanban_titles", "chain", lambda cs: _proc_chain(flow, cs)), + ("kanban_chains", "chain_verify", lambda cs: _proc_chain_verify(ctx, flow, cs)), + ("kanban_chains", "naming", lambda cs: _proc_naming(ctx, flow, cs)), + ("kanban_chains", "naming_verify", lambda cs: _proc_naming_verify(ctx, flow, cs)), + ("kanban_chains", "chain_filter", lambda cs: _proc_chain_filter(flow, cs)), + ("kanban_chains", "filter_verify", lambda cs: _proc_filter_verify(ctx, flow, cs)), + ("kanban_chains", "block_assemble", lambda cs: _proc_block(flow, cs)), + ("kanban_blocks", "small_blocks", lambda cs: _proc_small(ctx, flow, cs)), + ("kanban_blocks", "small_verify", lambda cs: _proc_small_verify(ctx, flow, cs)), + ("kanban_blocks", "dependency", lambda cs: _proc_dependency(ctx, flow, cs)), + ("kanban_blocks", "dependency_verify", lambda cs: _proc_dependency_verify(flow, cs)), + ("kanban_blocks", "main", lambda cs: _proc_main(flow, cs)), + ] + workers = [_research(ctx, files, q, folder, instructions, flow)] if research else [] + for table, stage, proc in specs: + workers.append(_worker(flow, table, stage, proc, up[stage], barrier=(stage in barriers), + inflight=(1 if stage in _SERIAL_STAGES else WORKER_INFLIGHT))) + + progress = asyncio.create_task(_progress(flow, set_p)) + try: + await asyncio.gather(*workers, return_exceptions=True) + finally: + flow.stop = True + progress.cancel() + _active_flows.pop(topic, None) + if ctx.is_cancelled(): + return False + n = await db.kanban_count(topic, "kanban_blocks", DONE_BLOCK) + _log(topic, f"Kanban: done — {n} blocks finalized") + return True + + +async def _progress(flow: _Flow, set_p): + while not flow.stop: + try: + counts = await db.kanban_stage_counts(flow.topic) + total = sum(counts.values()) + set_p(f"Kanban: {total} cards in flow") + except Exception: + pass + await asyncio.sleep(1.0) diff --git a/backend/models.py b/backend/models.py index 182638e..bb5c152 100644 --- a/backend/models.py +++ b/backend/models.py @@ -33,6 +33,7 @@ class BlocksCreateRequest(BaseModel): ab_phase: int | None = Field(default=None, ge=1, le=9) # re-run from a coarse phase (position in _phasen(topic), 1-based; up to 9: …outline/questions/artifacts); None = resume/continue without deleting ab_step: int | None = Field(default=None, ge=0) # re-run from a fine sub-step (0-based index into _blocks_steps); takes precedence over ab_phase to_step: int | None = Field(default=None, ge=0) # stop AFTER this fine sub-step (0-based index into _blocks_steps); None = run to the end + research: bool = True # False = "Continue": process the existing queue, search no new titles class BlocksResetStepRequest(BaseModel): diff --git a/backend/pipeline.py b/backend/pipeline.py index 5a2be44..02c4f8f 100644 --- a/backend/pipeline.py +++ b/backend/pipeline.py @@ -167,7 +167,7 @@ _yesno_schema = _enum_map_schema("relevant", _YESNO) # triage gate _MAX_RESTARTS = 2 -async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: int, provider: str, on_update=None, cancelled=None, *, grace: int | None = None) -> list | None: +async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: int, provider: str, on_update=None, cancelled=None, *, grace: int | None = None, min_runtime: int | None = None, max_runtime: int | None = None) -> list | None: """Starts all slots in parallel and collects `quorum` valid results. Slot spec: {key, prompt, role, capabilities, payload}. `payload(result)` @@ -180,10 +180,18 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: a timer of `grace` seconds. After it expires, running agents are only killed if the minimum stands — otherwise the race, including restarts, keeps running until it stands. Returns: `quorum` to `len(slots)` results. + + `min_runtime` (wall-clock from start): the race does not return before it + elapses while agents are still running — gives them time to search thoroughly. + `max_runtime` (wall-clock from start): hard cap — returns whatever is collected + (or None if nothing), killing the rest. Both default off; only Research sets them. """ attempts = {i: 0 for i in range(len(slots))} tasks: dict[asyncio.Task, int] = {} loop = asyncio.get_running_loop() + start = loop.time() + min_deadline = start + min_runtime if min_runtime else None + max_deadline = start + max_runtime if max_runtime else None deadline: float | None = None def spawn(i: int) -> None: @@ -191,7 +199,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, + scope=topic, on_line=slot.get("on_line"), )) tasks[task] = i @@ -203,12 +211,22 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: while tasks: if cancelled and cancelled(): return None - if deadline is not None and len(results) >= quorum and loop.time() >= deadline: + # Hard wall-clock cap: return whatever we have (None if empty), kill the rest. + if max_deadline is not None and loop.time() >= max_deadline: + _log(topic, f"{label}: max runtime {max_runtime}s reached ({len(results)} valid)") + return results or None + min_ok = min_deadline is None or loop.time() >= min_deadline + if deadline is not None and len(results) >= quorum and loop.time() >= deadline and min_ok: return results - # Grace set and minimum reached → only wait for the remaining deadline - wait_timeout = None + # Wake up for the earliest relevant deadline (grace, min, or max). + waits = [] if deadline is not None and len(results) >= quorum: - wait_timeout = max(0.0, deadline - loop.time()) + waits.append(deadline - loop.time()) + if min_deadline is not None: + waits.append(min_deadline - loop.time()) + if max_deadline is not None: + waits.append(max_deadline - loop.time()) + wait_timeout = max(0.0, min(waits)) if waits else None done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED, timeout=wait_timeout) if not done: continue @@ -235,7 +253,8 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: _log(topic, f"{label}: first result — grace {grace}s running") if on_update: on_update(len(results)) - if len(results) >= quorum and (grace is None or loop.time() >= deadline): + if (len(results) >= quorum and (grace is None or loop.time() >= deadline) + and (min_deadline is None or loop.time() >= min_deadline)): return results continue @@ -273,13 +292,13 @@ OK, CANCELLED, FAILED = "ok", "cancelled", "failed" async def run_single_slot( ctx: GenContext, label: str, *, - key: str, prompt: str, role: str, capabilities: str, payload, timeout: int, + key: str, prompt: str, role: str, capabilities: str, payload, timeout: int, on_line=None, ) -> tuple[str, object]: """One agent, one valid result (race with quorum 1). → (OK, value) | (CANCELLED, None) | (FAILED, None) """ - slots = [{"key": key, "prompt": prompt, "role": role, "capabilities": capabilities, "payload": payload}] + slots = [{"key": key, "prompt": prompt, "role": role, "capabilities": capabilities, "payload": payload, "on_line": on_line}] res = await _race(ctx.topic, label, slots, 1, timeout, ctx.provider, cancelled=ctx.is_cancelled) if ctx.is_cancelled(): return CANCELLED, None diff --git a/backend/routes.py b/backend/routes.py index 7404be1..2051cbf 100644 --- a/backend/routes.py +++ b/backend/routes.py @@ -164,7 +164,7 @@ async def create_blocks(req: BlocksCreateRequest): raise HTTPException(400, "Link must start with http:// or https://.") qp.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(qp, {"type": type, "location": location, "spec": req.instructions.strip()}) - asyncio.create_task(generate_blocks(topic, req.instructions.strip(), req.provider, ab_phase=req.ab_phase, ab_step=req.ab_step, to_step=req.to_step)) + asyncio.create_task(generate_blocks(topic, req.instructions.strip(), req.provider, ab_phase=req.ab_phase, ab_step=req.ab_step, to_step=req.to_step, research=req.research)) return {"ok": True} @@ -231,6 +231,28 @@ async def get_blocks_uebersicht(topic: str): return await load_overview(topic) +@router.get("/blocks/kanban") +async def get_kanban_board(topic: str): + """Live card counts per kanban column (empty dict when the streaming inventory is not in use).""" + import database as db + return await db.kanban_stage_counts(topic) + + +@router.get("/blocks/agents") +async def get_active_agents(topic: str): + """Currently running agents for this topic + their runtime (seconds). Label = key minus prefix.""" + from agents import active_agents + prefix = f"blocks-{topic}-" + return [{"label": a["key"][len(prefix):], "runtime": a["runtime"]} for a in active_agents(prefix)] + + +@router.post("/blocks/research") +async def add_research(topic: str): + """Attach one more research agent to the running kanban flow (live breadth boost).""" + from kanban import add_research_agent + return {"started": add_research_agent(topic)} + + @router.get("/blocks/question-pattern") async def get_question_pattern(topic: str, block: str): """Unlocked question patterns of a block (up to the current level; empty = live).""" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 6811578..1a37383 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -224,12 +224,12 @@ async function handleResetFromStep(step) { await loadBlocks() } -async function handleBlocksClick({ instructions, abPhase = null, abStep = null, toStep = null }) { +async function handleBlocksClick({ instructions, abPhase = null, abStep = null, toStep = null, research = true }) { if (!selectedTopic.value) return uiError.value = null try { // Source is already fixed here; abPhase/abStep set the start, toStep an optional end limit. - await apiCreateBausteine(selectedTopic.value, instructions, provider.value, undefined, undefined, abPhase, abStep, toStep) + await apiCreateBausteine(selectedTopic.value, instructions, provider.value, undefined, undefined, abPhase, abStep, toStep, research) } catch (e) { uiError.value = e.message return @@ -424,7 +424,7 @@ onMounted(async () => { @close="mainView = 'detail'" @restartFrom="(r) => handleBlocksClick({ instructions: '', abStep: r.from, toStep: r.to })" @resetFrom="handleResetFromStep" - @restartAll="() => handleBlocksClick({ abPhase: blocks.ready ? 1 : null })" + @restartAll="(o) => handleBlocksClick({ research: o?.research ?? false })" @removeAll="handleResetBlocks" @cancel="handleCancelBlocks" /> diff --git a/frontend/src/api.js b/frontend/src/api.js index 85ff9e4..3d03c7f 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -47,11 +47,11 @@ export async function fetchBlocksStatus(topic) { return res.json() } -export async function createBlocks(topic, instructions = '', provider = 'claude', sourceType = 'thema', sourceOrt = '', abPhase = null, abStep = null, toStep = null) { +export async function createBlocks(topic, instructions = '', provider = 'claude', sourceType = 'thema', sourceOrt = '', abPhase = null, abStep = null, toStep = null, research = true) { const res = await fetch(`${BASE}/blocks`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ topic, instructions, provider, source_type: sourceType, source_location: sourceOrt, ab_phase: abPhase, ab_step: abStep, to_step: toStep }), + body: JSON.stringify({ topic, instructions, provider, source_type: sourceType, source_location: sourceOrt, ab_phase: abPhase, ab_step: abStep, to_step: toStep, research }), }) return jsonOrThrow(res) } @@ -147,6 +147,24 @@ export async function fetchBlocksOverview(topic) { return jsonOrThrow(res) } +// Live card counts per kanban column ({} when the streaming inventory is not in use). +export async function fetchKanban(topic) { + const res = await fetch(`${BASE}/blocks/kanban?topic=${encodeURIComponent(topic)}`) + return jsonOrThrow(res) +} + +// Currently running agents for a topic + their runtime in seconds. +export async function fetchAgents(topic) { + const res = await fetch(`${BASE}/blocks/agents?topic=${encodeURIComponent(topic)}`) + return jsonOrThrow(res) +} + +// Attach one more research agent to the running kanban flow. +export async function addResearch(topic) { + const res = await fetch(`${BASE}/blocks/research?topic=${encodeURIComponent(topic)}`, { method: 'POST' }) + return jsonOrThrow(res) +} + export async function cancelGuide(id) { await fetch(`${BASE}/guides/${id}/cancel`, { method: 'POST' }) } diff --git a/frontend/src/components/BlocksOverview.vue b/frontend/src/components/BlocksOverview.vue index 8b084cb..2cda750 100644 --- a/frontend/src/components/BlocksOverview.vue +++ b/frontend/src/components/BlocksOverview.vue @@ -1,6 +1,6 @@