update

backend/board_inventory.py

@@ -14,6 +14,7 @@ Stages (cards):
   naming          cluster  judge picks the canonical member title
   naming_check    cluster  second judge verifies → spawns the block card
   fragment_filter block    BARRIER/drain: global re-merge + degrade pass (full list)
+  dedup           block    BARRIER/drain: global judge-verified pair dedup (incl. context)
   grouping        block    BARRIER/drain: umbrella grouping (type gate, reconcile)
   gap_check       block    BARRIER/drain: one supplement round (web) → feeds ingest
   done            block    mirror into the legacy `blocks` table → done_block
@@ -34,7 +35,7 @@ import kanban
 from kanban import Flow, Stage, chain_stages
 import blocks
 from blocks import (
-    DEDUP_PAIR_FLOOR, DEDUP_PAIRS_CHUNK, DEDUP_TITLE_AUTO, FILTER_CHUNK,
+    DEDUP_GLOBAL_FLOOR, DEDUP_PAIR_FLOOR, DEDUP_PAIRS_CHUNK, DEDUP_TITLE_AUTO, FILTER_CHUNK,
     FILTER_RECHECK_PANEL, CONSOLIDATION_PANEL, RESEARCH_BATCH, RESEARCH_READERS,
     RESEARCH_THEMA_AGENTS, _FILTER_NOTATION, _GROUP_STANDALONE,
     _build_research_prompt, _canonical, _canonical_key, _chunk_nums, _cliques,
@@ -425,7 +426,7 @@ async def _proc_consensus_gate(ctx: GenContext, flow: Flow, cards):
         rep = _rep(rows)
         p = c["payload"]
         p.update(title=rep["title"], description=rep["description"], readers=sorted(readers),
-                 supplement=supplement)
+                 supplement=supplement, n_size=len(readers))  # LPT: big evidence first
         if supplement or len(readers) >= 2:
             if _is_reference(rep["title"]) and not supplement:
                 p["quorum"] = "majority"  # consensus reference title: rename/exam, not the hard bar
@@ -598,7 +599,7 @@ async def _namecheck_one(ctx: GenContext, flow: Flow, c):
     sources = sorted(set().union(*[set(r.get("sources") or []) for r in rows])) if rows else []
     await db.kanban_upsert_card(topic, BOARD, f"b-{cid}", "block", "fragment_filter", {
         "title": p.get("title", ""), "description": p.get("description", ""),
-        "readers": readers, "sources": sources,
+        "readers": readers, "sources": sources, "n_size": len(readers),
         "supplement": bool(p.get("supplement")), "cluster": cid,
     })
     await db.kanban_advance(topic, BOARD, cid, "done_cluster")
@@ -864,7 +865,162 @@ async def _proc_fragment_filter(ctx: GenContext, flow: Flow, cards):
                            "floor_veto": [allrows[nr - 1]["title"] for nr in floor_veto]}, indent=1)
         _log(topic, f"Fragment-Filter: {n_dem} → {n_dem - len(journal)} (−{len(journal)})")
     demoted = {cid for cid, _ in moves}
-    moves += [(r["card_id"], "grouping") for r in rows if r["card_id"] not in demoted]
+    moves += [(r["card_id"], "dedup") for r in rows if r["card_id"] not in demoted]
+    await db.kanban_advance_many(topic, BOARD, moves)
+    flow.wake.set()
+
+
+async def _proc_dedup(ctx: GenContext, flow: Flow, cards):
+    """BARRIER/drain — global pair dedup over the NAMED blocks: filter's re-merge only
+    catches key-exact/≥0.95 titles inside the stage; here embedding candidates (mean OR
+    title cosine ≥ DEDUP_GLOBAL_FLOOR) + canonical-key blocking go to a TWO-judge panel
+    (merge needs unanimity — single judges conflate variants with their base entity),
+    auto edges (title ≥0.95, exact key) with relation guard, complete-link cliques merge
+    into the champion (readers/sources union). The second wave (supplement) compares
+    against the already-confirmed context blocks. Journal carries every pair verdict."""
+    topic = flow.topic
+    work_dir = flow.work_dir
+    rows = [{"card_id": c["card_id"], "payload": c["payload"],
+             "title": c["payload"].get("title", ""),
+             "description": c["payload"].get("description") or ""} for c in cards]
+    context = await _context_blocks(topic, exclude={r["card_id"] for r in rows})
+    allrows = rows + context
+    n_dem, n_all = len(rows), len(allrows)
+
+    async def _pass_through():
+        await db.kanban_advance_many(topic, BOARD, [(r["card_id"], "grouping") for r in rows])
+        flow.wake.set()
+
+    if n_all < 2 or not await _emb_ok(flow):
+        await _pass_through()
+        return
+    vf = await _vec_rows(flow, [_t_text(r) for r in allrows])
+    # titles casefolded: ALL-CAPS variants ("VERTEX COVER" vs "Vertex Cover (VC)") tank the
+    # raw title cosine below the candidate floor
+    vt = await _vec_rows(flow, [r["title"].casefold() for r in allrows])
+    if vf is None or vt is None:
+        await _pass_through()
+        return
+    sims_title = vt @ vt.T
+    sims = (vf @ vf.T + sims_title) / 2
+
+    def _demotable_pair(a: int, b: int) -> bool:
+        return a < n_dem or b < n_dem  # confirmed context blocks never merge among themselves
+
+    pairs: set[tuple[int, int]] = set()
+    for i in range(n_all):
+        for j in range(i + 1, n_all):
+            # title-only cosine as second candidate source: descriptions of the same
+            # entity often stress different facets and dilute the mean below the floor
+            if _demotable_pair(i, j) and (float(sims[i][j]) >= DEDUP_GLOBAL_FLOOR
+                                          or float(sims_title[i][j]) >= DEDUP_GLOBAL_FLOOR):
+                pairs.add((i, j))
+    keys: dict[str, list[int]] = {}
+    for i, r in enumerate(allrows):
+        if (k := _canonical_key(r["title"])):
+            keys.setdefault(k, []).append(i)
+    for grp in keys.values():
+        for x in range(len(grp)):
+            for y in range(x + 1, len(grp)):
+                if _demotable_pair(grp[x], grp[y]):
+                    pairs.add((grp[x], grp[y]))
+    ordered = sorted(pairs)
+    h = _h(*[r["card_id"] for r in allrows])
+    if not ordered:
+        atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
+                          {"vorher": n_dem, "paare": 0, "merged": []}, indent=1)
+        await _pass_through()
+        return
+    chunks = [ordered[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(ordered), DEDUP_PAIRS_CHUNK)]
+
+    async def _judge(ci, chunk, jj):
+        path = work_dir / f"dedup-{h}-c{ci}-j{jj}.json"
+        if _pairs_schema(_json_file(path)) is not None:
+            return  # resume
+        lines = "\n\n".join(f"{j + 1}.\nA: {_t_text(allrows[a])}\nB: {_t_text(allrows[b])}"
+                            for j, (a, b) in enumerate(chunk))
+        status, _v = await run_single_slot(
+            ctx, f"Dedup {ci} j{jj}", key=f"blocks-{topic}-dedup-{h}-c{ci}-j{jj}",
+            prompt=_prompt("Blocks-Dedup", topic=topic, pairs=lines, out_path=path),
+            role="judge", capabilities="files",
+            payload=lambda result, p=path: _pairs_schema(_json_file(p)),
+            timeout=_timeout("selection_mapping", len(chunk)))
+        if status == FAILED:
+            raise RuntimeError(f"Dedup chunk {ci} j{jj} ohne Ergebnis")
+
+    # two-judge panel per chunk (one wave): a merge needs UNANIMITY — the observed
+    # failure mode is a single judge conflating a variant with its base entity
+    results = await asyncio.gather(*[_judge(ci, c, jj) for ci, c in enumerate(chunks)
+                                     for jj in (1, 2)], return_exceptions=True)
+    errs = [r for r in results if isinstance(r, Exception)]
+    if errs:
+        raise errs[0]
+    if ctx.is_cancelled():
+        return
+    edges: list[tuple[int, int]] = []
+    detail: dict[tuple[int, int], str] = {}
+
+    def _edge(a, b, kanal):
+        if _relation_conflict(allrows[a]["title"], allrows[b]["title"]):
+            detail[(a, b)] = "guard_veto"
+        else:
+            edges.append((a, b))
+            detail[(a, b)] = kanal
+
+    for ci, chunk in enumerate(chunks):
+        v1 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j1.json")) or {}
+        v2 = _pairs_schema(_json_file(work_dir / f"dedup-{h}-c{ci}-j2.json")) or {}
+        for j, (a, b) in enumerate(chunk):
+            ja1, ja2 = bool(v1.get(j + 1)), bool(v2.get(j + 1))
+            if ja1 and ja2:
+                _edge(a, b, "ja")
+            else:
+                detail[(a, b)] = "nein" if not (ja1 or ja2) else "uneinig"
+    for a, b in ordered:  # auto recall net: near-identical titles merge without the judges
+        if float(sims_title[a][b]) >= DEDUP_TITLE_AUTO:
+            _edge(a, b, "auto_titel")
+    for grp in keys.values():  # exact-canonical-key auto-merge
+        for x in range(len(grp)):
+            for y in range(x + 1, len(grp)):
+                if _demotable_pair(grp[x], grp[y]):
+                    _edge(grp[x], grp[y], "auto_key")
+    moves: list[tuple[str, str]] = []
+    journal: list[dict] = []
+    gone: set[int] = set()
+    for g in _cliques(n_all, edges):
+        ctx_members = [k for k in g if k >= n_dem]
+        if ctx_members:
+            rep = ctx_members[0]  # confirmed block always wins
+        else:
+            rep = max(g, key=lambda k: (-_aspect_marker(allrows[k]["title"]),
+                                        len(allrows[k]["description"]), len(allrows[k]["title"]), -k))
+        rp = allrows[rep]["payload"]
+        changed = False
+        for k in g:
+            if k == rep or k >= n_dem:
+                continue  # context members stay untouched
+            lp = allrows[k]["payload"]
+            rp["readers"] = sorted(set(rp.get("readers") or []) | set(lp.get("readers") or []))
+            rp["sources"] = sorted(set(rp.get("sources") or []) | set(lp.get("sources") or []))
+            lp.update(reason="merged", merged_into=allrows[rep]["title"])
+            await db.kanban_set_payload(topic, BOARD, allrows[k]["card_id"], lp)
+            moves.append((allrows[k]["card_id"], "grouped"))
+            gone.add(k)
+            journal.append({"dublette": allrows[k]["title"], "in": allrows[rep]["title"]})
+            changed = True
+        if changed:
+            await db.kanban_set_payload(topic, BOARD, allrows[rep]["card_id"], rp)
+            if rep >= n_dem and rp.get("mirrored_norm"):  # mirror refresh (sources only)
+                await db.upsert_block(topic, rp["mirrored_norm"], allrows[rep]["title"],
+                                      allrows[rep]["description"], rp.get("sources") or [])
+    moves += [(r["card_id"], "grouping") for i, r in enumerate(rows) if i not in gone]
+    atomic_write_json(work_dir / f"inventar-dedup-{h}.json",
+                      {"vorher": n_dem, "paare": len(ordered), "merged": journal,
+                       "paare_detail": [{"a": allrows[a]["title"], "b": allrows[b]["title"],
+                                         "verdict": v} for (a, b), v in sorted(detail.items())]},
+                      indent=1)
+    if journal:
+        _log(topic, f"Dedup: {len(journal)} Dublette(n) zusammengelegt")
     await db.kanban_advance_many(topic, BOARD, moves)
     flow.wake.set()
 
@@ -1202,6 +1358,8 @@ def inventory_stages(ctx: GenContext, flow: Flow) -> list[Stage]:
         Stage(BOARD, "naming_check", lambda cs: _proc_naming_check(ctx, flow, cs)),
         Stage(BOARD, "fragment_filter", lambda cs: _proc_fragment_filter(ctx, flow, cs),
               barrier=True, drain=True, gate=research_done),
+        Stage(BOARD, "dedup", lambda cs: _proc_dedup(ctx, flow, cs),
+              barrier=True, drain=True, gate=research_done),
         Stage(BOARD, "grouping", lambda cs: _proc_grouping(ctx, flow, cs),
               barrier=True, drain=True, gate=research_done),
         Stage(BOARD, "gap_check", lambda cs: _proc_gap_check(ctx, flow, cs),
@@ -1305,6 +1463,7 @@ COLUMNS = [
     ("inventory", "naming", "Naming", "cluster"),
     ("inventory", "naming_check", "Naming-Check", "cluster"),
     ("inventory", "fragment_filter", "Fragment-Filter", "block"),
+    ("inventory", "dedup", "Dubletten", "block"),
     ("inventory", "grouping", "Gruppierung", "block"),
     ("inventory", "gap_check", "Lücken-Check", "block"),
     ("inventory", "done", "Spiegeln", "block"),
@@ -1324,7 +1483,7 @@ COLUMNS = [
 
 _TITLE_STAGES = ["ingest", "cluster"]
 _CLUSTER_STAGES = ["pair_check", "consensus_gate", "clarify", "naming", "naming_check"]
-_BLOCK_STAGES = ["fragment_filter", "grouping", "gap_check", "done"]
+_BLOCK_STAGES = ["fragment_filter", "dedup", "grouping", "gap_check", "done"]
 _ART_STAGES = ["subblocks", "facts", "levels", "relevance", "question_pattern",
                "artefacts", "finalize", "outline"]
 # where a requeued dead card restarts, by kind