fix(dedup): Intra-Run-Sibling-Dedup (Befund D)

generative/orchestrator.py

@@ -831,6 +831,145 @@ def dedup_exact(drafts: list[AtomicNoteDraft],
     return result
 
 
+def resolve_sibling_dups(drafts: list[AtomicNoteDraft],
+                         existing_concepts: dict[str, str] | None = None
+                         ) -> tuple[list[AtomicNoteDraft], int]:
+    """Intra-Run-Sibling-Dedup (Befund D).
+
+    cross_reference erkennt zwei Near-Dup-Drafts EINES Laufs (dup_risk=high) und setzt
+    action=extend + extend_path=<Sibling-Titel>. Da der Sibling keine Vault-Datei ist,
+    verpufft das Signal beim Writer (write_note routet nur über find_existing_in_vault,
+    nicht über extend_path) und BEIDE Notes werden als Vollnoten geschrieben. Diese
+    Funktion wertet genau dieses bereits gesetzte extend-Signal aus und kollabiert solche
+    Geschwister deterministisch zu EINER Note — VOR dem Schreiben.
+
+    Bewusst KEIN Eingriff ins Title-Blocking von entity_resolution: hier wird nur das vom
+    LLM bereits gefällte Dup-Urteil interpretiert, kein neuer Body-Cosine-Pass (der echte,
+    distinkt betitelte Geschwister fälschlich mergen könnte) und kein zusätzlicher LLM-Call.
+
+    Survivor pro Cluster: höchster critic_score, Tie → längerer Body, Tie → norm-Titel
+    (ordnungsunabhängig deterministisch). Verlustarm: related-Links + source_anchors der
+    gedroppten Drafts wandern in den Survivor (related auf MAX_RELATED gedeckelt); gedroppte
+    Titel/Aliase werden Survivor-Aliase, sodass [[…]]-Links auf den gedroppten Titel auf den
+    Survivor auflösen (kein Dead-Link).
+
+    Vault-Erhalt (Cross-Model-Review Codex 2026-06-23, HIGH#2): hat IRGENDEIN Cluster-Member
+    eine Vault-Dublette (action=extend mit extend_path auf eine reale Vault-Note), erbt der
+    Survivor diesen Bezug (action=extend + Vault-Stem als Alias → title-/alias-basierter
+    Writer findet die Vault-Note). Sonst wird ein dangling Intra-Run-extend auf 'create'
+    zurückgesetzt.
+
+    Präkondition: dedup_exact lief vorher → alle Drafts haben unique normalisierte Titel
+    (keine norm-Title-Kollisionen, Codex MED#3).
+    """
+    from generative.agents.cross_reference import MAX_RELATED
+    n = len(drafts)
+    if n <= 1:
+        return drafts, 0
+
+    norm_title = [_normalize(d.title) for d in drafts]
+    title_to_idx = {nt: i for i, nt in enumerate(norm_title)}
+
+    # Vault-Index normalisieren (Keys = Titel, Values = Pfade) — für Vault-Dubletten-Erkennung
+    ec = existing_concepts or {}
+    vault_norms = {_normalize(k) for k in ec} | {_normalize(Path(v).stem) for v in ec.values()}
+
+    def _vault_target(ep: str | None) -> bool:
+        """True wenn extend_path auf eine reale Vault-Note zeigt (Titel- ODER Stem-Match)."""
+        if not ep:
+            return False
+        return _normalize(ep) in vault_norms or _normalize(Path(ep).stem) in vault_norms
+
+    def _match_idx(ep: str, self_i: int) -> int | None:
+        """In-Run-Draft, dessen Titel zu extend_path passt — per direkt-normalize ODER
+        Path-Stem (Codex MED#4: bare Titel mit '/' dürfen nicht über stem zerlegt werden)."""
+        for key in (_normalize(ep), _normalize(Path(ep).stem)):
+            j = title_to_idx.get(key)
+            if j is not None and j != self_i:
+                return j
+        return None
+
+    # Union-Find über Intra-Run-extend-Kanten (gleiches Cluster-Pattern wie entity_resolution)
+    parent = list(range(n))
+
+    def find(x: int) -> int:
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    def union(a: int, b: int) -> None:
+        ra, rb = find(a), find(b)
+        if ra != rb:
+            parent[rb] = ra
+
+    for i, d in enumerate(drafts):
+        if d.action != "extend" or not d.extend_path:
+            continue
+        j = _match_idx(d.extend_path, i)
+        if j is not None:  # extend_path trifft einen anderen In-Run-Draft → Geschwister
+            union(i, j)
+
+    clusters: dict[int, list[int]] = {}
+    for i in range(n):
+        clusters.setdefault(find(i), []).append(i)
+
+    def _link_norm(link: str) -> str:
+        s = link.strip()
+        if s.startswith("[[") and s.endswith("]]"):
+            s = s[2:-2]
+        s = s.split("|", 1)[0].split("#", 1)[0].strip()  # Alias- und Heading-Anker abtrennen
+        return _normalize(Path(s).stem)
+
+    drop_idx: set[int] = set()
+    for members in clusters.values():
+        if len(members) <= 1:  # Multi-Member-Cluster entsteht nur durch eine Sibling-Kante
+            continue
+        survivor = max(members, key=lambda m: (drafts[m].critic_score,
+                                               len(drafts[m].body or ""),
+                                               norm_title[m]))
+        s = drafts[survivor]
+        alias_norms = {_normalize(a) for a in s.aliases} | {norm_title[survivor]}
+
+        def _absorb_alias(name: str) -> None:
+            if name and _normalize(name) not in alias_norms:
+                s.aliases.append(name)
+                alias_norms.add(_normalize(name))
+
+        for m in members:
+            if m == survivor:
+                continue
+            d = drafts[m]
+            drop_idx.add(m)
+            for alias in [d.title, *d.aliases]:
+                _absorb_alias(alias)
+            s.source_anchors.extend(d.source_anchors)
+            for link in d.related:
+                if link not in s.related:
+                    s.related.append(link)
+
+        # Vault-Erhalt: hat ein Cluster-Member eine reale Vault-Dublette, erbt der Survivor
+        # sie. Das eigene Vault-Ziel des Survivors hat Vorrang (Mistral-Review 2026-06-23),
+        # sonst der erste Member in Index-Reihenfolge.
+        _vault_order = [survivor] + [m for m in sorted(members) if m != survivor]
+        vault_ep = next((drafts[m].extend_path for m in _vault_order
+                         if drafts[m].action == "extend" and _vault_target(drafts[m].extend_path)),
+                        None)
+        if vault_ep is not None:
+            s.action = "extend"
+            s.extend_path = vault_ep
+            _absorb_alias(Path(vault_ep).stem)  # Writer findet Vault-Note via Alias
+        elif s.action == "extend" and s.extend_path:  # dangling Intra-Run-extend
+            s.action = "create"
+            s.extend_path = None
+
+        # Self-Links (auf Survivor-Titel oder absorbierte Aliase) entfernen, dann deckeln
+        s.related = [l for l in s.related if _link_norm(l) not in alias_norms][:MAX_RELATED]
+
+    kept = [d for i, d in enumerate(drafts) if i not in drop_idx]
+    return kept, len(drop_idx)
+
+
 def _auto_start_dashboard() -> None:
     """Startet den Dashboard-Server im Hintergrund falls er noch nicht läuft."""
     import socket, subprocess
@@ -1524,6 +1663,16 @@ def main(argv: list[str] | None = None):
             refine_budget=refine_budget,
         ))
 
+    # --- Dedup Stage C: Intra-Run-Sibling-Dedup (Befund D) ---
+    # cross_reference setzt bei dup_risk=high action=extend + extend_path=<Sibling-Titel>.
+    # Zeigt das auf einen Draft DESSELBEN Laufs (keine Vault-Datei), verpufft es beim
+    # Writer und beide Notes würden geschrieben. Hier auf das vorhandene Signal reagieren
+    # und Geschwister eines Laufs deterministisch zu EINER Note kollabieren — nach den
+    # per-Draft-Calls (Signal steht erst jetzt fest), vor boilerplate_dedup und Writer.
+    drafts, n_sib = resolve_sibling_dups(drafts, existing_concepts)
+    if n_sib:
+        print(f"      [sibling-dedup] {n_sib} Intra-Run-Near-Dup(s) in Geschwister-Note(s) gemergt")
+
     # --- Hebel #5: Boilerplate-Dedup zwischen Hub-Drafts und Sub-Konzept-Drafts ---
     drafts, stripped = boilerplate_dedup.dedup_hub_subconcepts(drafts)
     if stripped:

generative/tests/test_sibling_dedup.py

@@ -0,0 +1,204 @@
+"""Tests für Intra-Run-Sibling-Dedup (Befund D).
+
+cross_reference erkennt zwei Near-Dup-Drafts EINES Laufs (dup_risk=high) und setzt
+action=extend + extend_path=<Sibling-Titel>. Da der Sibling keine Vault-Datei ist,
+verpufft das beim Writer und BEIDE Notes werden geschrieben. resolve_sibling_dups()
+wertet genau dieses vorhandene Signal aus und mergt/skippt die Siblings VOR dem
+Schreiben — ohne Eingriff ins Title-Blocking (kein False-Positive-Risiko).
+"""
+from generative.orchestrator import resolve_sibling_dups
+from generative.agents.cross_reference import MAX_RELATED
+from generative.pipeline.vault_writer import write_note
+from generative.schemas.atomic_note import AtomicNoteDraft, TextAnchor
+
+
+def _draft(title, *, body="", action="create", extend_path=None,
+           related=None, source_anchors=None, critic_score=0, aliases=None):
+    return AtomicNoteDraft(
+        title=title,
+        body=body or f"Body von {title}",
+        source_anchors=source_anchors or [],
+        related=related or [],
+        tags=[],
+        synthesis_confidence="high",
+        aliases=aliases or [],
+        action=action,
+        extend_path=extend_path,
+        critic_score=critic_score,
+    )
+
+
+def test_pair_extend_to_sibling_merges_to_one():
+    # d_b ist ein Near-Dup von d_a und zeigt per extend_path auf dessen Titel.
+    d_a = _draft("Affective Access", body="Langer verifizierter Body " * 5,
+                 critic_score=4, related=["[[Information Behavior]]"])
+    d_b = _draft("Affektiver Zugang", action="extend",
+                 extend_path="Affective Access", critic_score=2,
+                 related=["[[Kuhlthau ISP]]"])
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 1
+    assert len(kept) == 1
+    survivor = kept[0]
+    assert survivor.title == "Affective Access"        # höherer critic_score + längerer Body
+    assert survivor.action == "create"                  # kein dangling extend
+    # related des gedroppten Drafts wandern verlustarm in den Survivor
+    assert "[[Kuhlthau ISP]]" in survivor.related
+    # gedroppter Titel lebt als Alias weiter → [[Affektiver Zugang]] löst auf den Survivor auf
+    assert "Affektiver Zugang" in survivor.aliases
+
+
+def test_cycle_a_to_b_and_b_to_a_resolves_to_one():
+    # Beide Drafts flaggen sich gegenseitig als Dup (paralleler per-Draft-Call).
+    d_a = _draft("A", action="extend", extend_path="B", critic_score=3)
+    d_b = _draft("B", action="extend", extend_path="A", critic_score=1)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 1
+    assert len(kept) == 1
+    assert kept[0].title == "A"                          # höherer critic_score gewinnt
+    assert kept[0].action == "create"                   # Zyklus aufgelöst, kein dangling extend
+
+
+def test_chain_a_b_c_collapses_to_one():
+    # A→B, B→C: Union-Find muss die ganze Kette zu einem Cluster verbinden.
+    d_a = _draft("A", action="extend", extend_path="B", critic_score=1)
+    d_b = _draft("B", action="extend", extend_path="C", critic_score=5)
+    d_c = _draft("C", critic_score=2)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b, d_c])
+
+    assert dropped == 2
+    assert len(kept) == 1
+    assert kept[0].title == "B"                          # höchster critic_score
+
+
+def test_vault_extend_is_not_touched():
+    # extend_path zeigt auf eine echte Vault-Note (Stem matcht KEINEN Sibling-Titel)
+    # → legitimer Vault-Extend, darf NICHT angefasst werden.
+    d_a = _draft("Concept A", critic_score=3)
+    d_b = _draft("Concept B", action="extend",
+                 extend_path="04-wissen/Some Vault Note.md", critic_score=2)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 0
+    assert len(kept) == 2
+    survivor_b = next(d for d in kept if d.title == "Concept B")
+    assert survivor_b.action == "extend"
+    assert survivor_b.extend_path == "04-wissen/Some Vault Note.md"
+
+
+def test_related_union_capped_and_no_dangling_self_link():
+    # Survivor-related enthält bereits einen Link auf den gedroppten Titel
+    # (cross_reference fügt [[dup-stem]] ein). Nach dem Drop darf kein Link auf
+    # den gedroppten Titel als Dead-Link überleben (Alias-Auflösung übernimmt),
+    # und die Gesamtzahl ist auf MAX_RELATED gedeckelt.
+    d_a = _draft("Survivor", critic_score=5,
+                 related=["[[Affektiver Zugang]]", "[[L1]]", "[[L2]]"])
+    d_b = _draft("Affektiver Zugang", action="extend", extend_path="Survivor",
+                 critic_score=1, related=["[[L3]]", "[[L4]]", "[[L5]]"])
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 1
+    survivor = kept[0]
+    assert survivor.title == "Survivor"
+    assert len(survivor.related) <= MAX_RELATED
+    # kein Self-Link auf den absorbierten Titel
+    assert "[[Affektiver Zugang]]" not in survivor.related
+
+
+def test_no_extend_drafts_is_noop():
+    d_a = _draft("A", critic_score=1)
+    d_b = _draft("B", critic_score=2)
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+    assert dropped == 0
+    assert len(kept) == 2
+
+
+# --- Cross-Model-Review-Befunde (Codex 2026-06-23) ---
+
+def test_vault_extend_propagates_to_survivor():
+    # HIGH#2: A (bester Body) ist Near-Dup von B, B ist zugleich Dup einer EXISTIERENDEN
+    # Vault-Note V. Survivor A behält seinen besseren Body, MUSS aber B's Vault-Bezug erben
+    # — sonst wird eine Dublette der Vault-Note geschrieben. Vault-Stem wird Alias, damit der
+    # title-/alias-basierte Writer die Vault-Note findet.
+    d_a = _draft("A", body="Langer Body " * 10, action="extend",
+                 extend_path="B", critic_score=5)
+    d_b = _draft("B", action="extend", extend_path="Vault Concept", critic_score=1)
+    existing = {"vault concept": "01-studium/Vault Concept.md"}
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b], existing)
+
+    assert dropped == 1
+    survivor = kept[0]
+    assert survivor.title == "A"                         # besserer Body bleibt Survivor
+    assert survivor.action == "extend"                   # Vault-Bezug NICHT verloren
+    assert survivor.extend_path == "Vault Concept"
+    assert any("vault concept" == a.lower() for a in survivor.aliases)  # Writer findet Vault-Note
+
+
+def test_bare_title_with_slash_matches():
+    # MED#4: Ein Sibling-Titel mit "/" darf nicht über Path().stem zerlegt werden.
+    d_a = _draft("TCP/IP", critic_score=4)
+    d_b = _draft("TCP IP Stack", action="extend", extend_path="TCP/IP", critic_score=1)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 1
+    assert kept[0].title == "TCP/IP"
+
+
+def test_survivor_tiebreak_is_order_independent():
+    # LOW#5: bei gleichem critic_score UND gleicher Body-Länge muss der Survivor
+    # deterministisch sein, unabhängig von der Eingabereihenfolge.
+    def make_pair():
+        a = _draft("Alpha", body="x" * 50, action="extend", extend_path="Beta", critic_score=3)
+        b = _draft("Beta", body="y" * 50, action="extend", extend_path="Alpha", critic_score=3)
+        return a, b
+
+    a1, b1 = make_pair()
+    kept1, _ = resolve_sibling_dups([a1, b1])
+    a2, b2 = make_pair()
+    kept2, _ = resolve_sibling_dups([b2, a2])  # umgekehrte Reihenfolge
+
+    assert kept1[0].title == kept2[0].title
+
+
+def test_heading_self_link_removed():
+    # LOW#6: [[Titel#Abschnitt]]-Self-Link auf den absorbierten Titel muss entfernt werden.
+    d_a = _draft("Survivor", critic_score=5,
+                 related=["[[Affektiver Zugang#Definition]]", "[[Keep]]"])
+    d_b = _draft("Affektiver Zugang", action="extend", extend_path="Survivor", critic_score=1)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b])
+
+    assert dropped == 1
+    assert not any("affektiver zugang" in l.lower() for l in kept[0].related)
+    assert "[[Keep]]" in kept[0].related
+
+
+def test_e2e_two_near_dups_write_one_note(tmp_path):
+    # Akzeptanzkriterium e2e (deterministisch): zwei Near-Dup-Drafts EINES Laufs mit dem
+    # realen cross_reference-Signal (B.action=extend, extend_path=<A-Titel>) ergeben nach
+    # resolve_sibling_dups + Writer GENAU EINE Note-Datei (ohne Fix wären es zwei).
+    anchors = [TextAnchor(quote="Webinare wirken vergleichbar", page="S. 5")]
+    d_a = _draft("Webinar-Wirksamkeit", body="Webinare zeigen vergleichbare Lerneffekte. " * 4,
+                 source_anchors=list(anchors), critic_score=4)
+    d_b = _draft("Wirksamkeit von Webinaren", body="Webinare sind ähnlich wirksam. " * 3,
+                 source_anchors=list(anchors), action="extend",
+                 extend_path="Webinar-Wirksamkeit", critic_score=2)
+
+    kept, dropped = resolve_sibling_dups([d_a, d_b], existing_concepts={})
+    assert dropped == 1 and len(kept) == 1
+
+    src = "Ebner und Gegenfurtner - 2019.pdf"
+    meta = {"Author": "Ebner & Gegenfurtner", "Year": "2019", "Title": "Webinar Meta-Analysis"}
+    for note in kept:
+        write_note(note, src, source_meta=meta, existing_concepts={}, inbox_dir=tmp_path)
+
+    written = list(tmp_path.glob("*.md"))
+    assert len(written) == 1, f"erwartet 1 Note, geschrieben: {[p.name for p in written]}"