From f8b3768c297d91f18d006ee31be9b43f7ec50b37 Mon Sep 17 00:00:00 2001
From: TillQuandel <tillq@live.de>
Date: Sat, 18 Jul 2026 16:35:19 +0200
Subject: [PATCH] feat(orchestrator): Kandidaten-Dedup + balancierter Cap fuer
 Hybrid-Buchplanung (#346, PR 2/3)

---
 generative/orchestrator.py                    | 176 ++++++++++++
 .../tests/test_candidate_dedup_and_cap.py     | 258 ++++++++++++++++++
 2 files changed, 434 insertions(+)
 create mode 100644 generative/tests/test_candidate_dedup_and_cap.py

diff --git a/generative/orchestrator.py b/generative/orchestrator.py
index c6c072b..0a77a74 100644
--- a/generative/orchestrator.py
+++ b/generative/orchestrator.py
@@ -1377,6 +1377,182 @@ def dedup_exact(drafts: list[AtomicNoteDraft], existing_concepts: dict[str, str]
     return result
 
 
+# --- Hybrid-Buchplanung: globale Kandidaten-Dedup + balancierter Cap (#346) --
+# Zwei reine, orchestrator-lokale Bausteine. Verdrahtung (book-mode) folgt in
+# PR 3; hier bewusst noch ohne Aufrufer (isolierte Review-Fläche). Beide sind
+# duck-typed auf ConceptItem-artige Objekte (title/priority/action/origin via
+# getattr) und teilen die _normalize()-Titelnormalisierung mit dedup_exact.
+
+_PRIORITY_RANK = {"high": 2, "medium": 1, "low": 0}
+
+
+def _priority_rank(concept) -> int:
+    """Prioritäts-Rang high>medium>low; unbekannt/None → 0 (schwächste Fassung)."""
+    return _PRIORITY_RANK.get(getattr(concept, "priority", None) or "", 0)
+
+
+def _candidate_beats(new, survivor) -> bool:
+    """True, wenn `new` die stärkere Fassung eines Titel-Duplikats ist:
+    höhere Priorität schlägt niedrigere; bei gleicher Priorität schlägt eine
+    non-skip-Fassung einen skip-Survivor; sonst gewinnt das Erstauftreten."""
+    rn, rs = _priority_rank(new), _priority_rank(survivor)
+    if rn != rs:
+        return rn > rs
+    return getattr(survivor, "action", None) == "skip" and getattr(new, "action", None) != "skip"
+
+
+def dedup_concept_candidates(concepts: list) -> tuple[list, int]:
+    """Globale Kandidaten-Dedup über den normalisierten Titel (#346).
+
+    Muster analog zu dedup_exact, aber für Planner-Konzept-Kandidaten aus
+    mehreren Kapiteln: dasselbe Konzept aus zwei Kapiteln wird EIN Konzept.
+
+    - origin=="secondary_mention" ist ein eigener Kanal: passiert UNANGETASTET,
+      wird nie dedupliziert oder verworfen und kollidiert nicht mit primären
+      Titeln.
+    - Bei Titel-Duplikat gewinnt die stärkere Fassung (siehe _candidate_beats)
+      und behält die Erstauftreten-Position (stabile Reihenfolge).
+    - Je verworfenem Duplikat ein Funnel-Event
+      (stage=planner_dedup, drop_reason=chapter_duplicate, detail=survivor=…).
+
+    Rückgabe: (deduplizierte Liste in stabiler Reihenfolge, Anzahl Duplikate).
+    """
+    result: list = []
+    pos_by_key: dict[str, int] = {}
+    dropped = 0
+    for c in concepts:
+        if getattr(c, "origin", None) == "secondary_mention":
+            result.append(c)
+            continue
+        key = _normalize(c.title)
+        if key not in pos_by_key:
+            pos_by_key[key] = len(result)
+            result.append(c)
+            continue
+        pos = pos_by_key[key]
+        survivor = result[pos]
+        if _candidate_beats(c, survivor):
+            # Neue Fassung übernimmt die Erstauftreten-Position; alte wird verworfen.
+            _trace_stage_outcome(
+                survivor.title,
+                "planner_dedup",
+                "dropped",
+                drop_reason="chapter_duplicate",
+                detail=f"survivor={c.title}",
+            )
+            result[pos] = c
+        else:
+            _trace_stage_outcome(
+                c.title,
+                "planner_dedup",
+                "dropped",
+                drop_reason="chapter_duplicate",
+                detail=f"survivor={survivor.title}",
+            )
+        dropped += 1
+    return result, dropped
+
+
+def _select_balanced(
+    pool_idx: list[int],
+    slots: int,
+    chapter_order: list,
+    chapter_keys: list,
+    chapter_word_counts: dict,
+    concepts: list,
+) -> set[int]:
+    """Wählt `slots` Kandidaten aus `pool_idx`, Slots proportional zum
+    Kapitel-Wortanteil (D'Hondt-Höchstzahlverfahren), innerhalb eines Kapitels
+    nach Priorität (desc) dann Erstauftreten. Deterministisch; Tie-Break Richtung
+    frühere Kapitel-Reihenfolge. Rückgabe: Menge gewählter Indizes."""
+    if slots <= 0:
+        return set()
+    by_chapter: dict = {}
+    for i in pool_idx:
+        by_chapter.setdefault(chapter_keys[i], []).append(i)
+    for members in by_chapter.values():
+        members.sort(key=lambda i: (-_priority_rank(concepts[i]), i))
+
+    chapters = [k for k in chapter_order if by_chapter.get(k)]
+    demand = {k: len(by_chapter[k]) for k in chapters}
+    to_fill = min(slots, sum(demand.values()))
+    alloc = {k: 0 for k in chapters}
+    idx_of = {k: n for n, k in enumerate(chapters)}
+    for _ in range(to_fill):
+        eligible = [k for k in chapters if alloc[k] < demand[k]]
+        # Höchstzahl weight/(alloc+1); Gleichstand -> frühere Kapitel-Reihenfolge.
+        best = max(
+            eligible,
+            key=lambda k: (chapter_word_counts.get(k, 0) / (alloc[k] + 1), -idx_of[k]),
+        )
+        alloc[best] += 1
+
+    selected: set[int] = set()
+    for k in chapters:
+        selected.update(by_chapter[k][: alloc[k]])
+    return selected
+
+
+def cap_candidates_balanced(
+    concepts: list,
+    budget: int,
+    chapter_word_counts: dict,
+    chapter_keys: list,
+) -> tuple[list, list]:
+    """Wortanteil-balancierter, prioritätssortierter Kandidaten-Cap (#346).
+
+    Sibling zu cap_actionable_concepts (runtime_config.py) für den book-mode-Pfad;
+    das bestehende Muster (uniform, keine Wortanteil-Balancierung) bleibt für
+    Normal-/by-chapter-Pfad unangetastet.
+
+    - `budget` ist fertig berechnet (Formel lebt beim Aufrufer, PR 3).
+    - `chapter_keys` ist index-aligned zu `concepts` (Herkunfts-Kapitel je
+      Kandidat). Bewusst paralleler Parameter statt ConceptItem.chapter: letzteres
+      ist LLM-Freitext ("wo erwartet"), nicht garantiert deckungsgleich mit den
+      deterministischen Outline-Keys von `chapter_word_counts`.
+    - Erst ALLE high-priority-Kandidaten über alle Kapitel; wenn highs > Budget,
+      werden sie proportional zum Wortanteil des Herkunfts-Kapitels gekürzt.
+    - Rest-Slots nach Kapitel-Wortanteil verteilt (großes Kapitel → mehr Slots),
+      innerhalb eines Kapitels nach Priorität, dann Erstauftreten.
+    - origin=="secondary_mention" zählt nicht gegen das Budget und wird nie
+      gekappt.
+
+    Rückgabe: (behaltene, gekappte) in Original-Eingabereihenfolge.
+    """
+    if len(chapter_keys) != len(concepts):
+        raise ValueError("chapter_keys muss index-aligned zu concepts sein")
+
+    budgeted_idx = [i for i, c in enumerate(concepts) if getattr(c, "origin", None) != "secondary_mention"]
+    if budget >= len(budgeted_idx):
+        return list(concepts), []  # no-op
+
+    # Kapitel-Reihenfolge deterministisch aus Erstauftreten der Kandidaten.
+    chapter_order: list = []
+    seen_ch: set = set()
+    for i in budgeted_idx:
+        k = chapter_keys[i]
+        if k not in seen_ch:
+            seen_ch.add(k)
+            chapter_order.append(k)
+
+    highs_idx = [i for i in budgeted_idx if _priority_rank(concepts[i]) == 2]
+    kept_idx: set[int] = set()
+    if len(highs_idx) > budget:
+        # Überlauf: highs proportional zum Wortanteil kürzen, keine Rest-Slots.
+        kept_idx |= _select_balanced(highs_idx, budget, chapter_order, chapter_keys, chapter_word_counts, concepts)
+    else:
+        kept_idx.update(highs_idx)
+        remaining = budget - len(highs_idx)
+        rest_idx = [i for i in budgeted_idx if _priority_rank(concepts[i]) != 2]
+        kept_idx |= _select_balanced(rest_idx, remaining, chapter_order, chapter_keys, chapter_word_counts, concepts)
+
+    kept = [c for i, c in enumerate(concepts) if i in kept_idx or getattr(c, "origin", None) == "secondary_mention"]
+    capped = [
+        c for i, c in enumerate(concepts) if i not in kept_idx and getattr(c, "origin", None) != "secondary_mention"
+    ]
+    return kept, capped
+
+
 def resolve_sibling_dups(
     drafts: list[AtomicNoteDraft], existing_concepts: dict[str, str] | None = None
 ) -> tuple[list[AtomicNoteDraft], int]:
diff --git a/generative/tests/test_candidate_dedup_and_cap.py b/generative/tests/test_candidate_dedup_and_cap.py
new file mode 100644
index 0000000..66a9cc9
--- /dev/null
+++ b/generative/tests/test_candidate_dedup_and_cap.py
@@ -0,0 +1,258 @@
+"""Tests für die isolierten Hybrid-Buchplanungs-Bausteine (#346, PR 2/3).
+
+Zwei reine Funktionen in orchestrator.py, die PR 3 später verdrahtet:
+
+- ``dedup_concept_candidates`` — globale Kandidaten-Dedup über normalisierten
+  Titel; stärkere Fassung (Priorität, dann non-skip) gewinnt; secondary_mention
+  ist ein eigener Kanal (nie dedupliziert); je verworfenem Duplikat ein
+  ``planner_dedup``-Funnel-Event.
+- ``cap_candidates_balanced`` — wortanteil-balancierter, prioritätssortierter
+  Kandidaten-Cap; highs zuerst über alle Kapitel, Überlauf proportional zum
+  Wortanteil gekürzt; Rest-Slots nach Wortanteil verteilt; secondary_mention
+  budget-neutral.
+
+Backend-/Capture-Muster wie test_stage_outcome_events.py: echtes JsonlBackend
+auf tmp umbiegen, JSONL zurücklesen.
+"""
+
+from __future__ import annotations
+
+import json
+
+import generative.agents.tracing as tracing
+from generative.agents.tracing import JsonlBackend
+from generative import orchestrator as orch
+from generative.schemas.atomic_note import ConceptItem
+
+
+# --- Helfer -----------------------------------------------------------------
+
+
+def _concept(
+    title: str,
+    priority: str = "medium",
+    chapter: str = "K1",
+    action: str = "create",
+    origin: str = "primary",
+) -> ConceptItem:
+    return ConceptItem(
+        title=title,
+        priority=priority,
+        chapter=chapter,
+        action=action,
+        origin=origin,
+    )
+
+
+def _capture(monkeypatch, tmp_path):
+    """Biegt das Trace-Backend auf tmp um; gibt einen Reader für stage_outcome-Events zurück."""
+    backend = JsonlBackend(run_dir=tmp_path, run_id="test-run")
+    monkeypatch.setattr(tracing, "_backend", backend)
+
+    def _read_stage_events() -> list[dict]:
+        f = tmp_path / "test-run.jsonl"
+        if not f.exists():
+            return []
+        events = [json.loads(line) for line in f.read_text(encoding="utf-8").splitlines() if line.strip()]
+        return [e for e in events if e.get("type") == "stage_outcome"]
+
+    return _read_stage_events
+
+
+# ===========================================================================
+# dedup_concept_candidates
+# ===========================================================================
+
+
+def test_dedup_single_occurrence_no_drop():
+    c = _concept("Data Governance")
+    kept, dropped = orch.dedup_concept_candidates([c])
+    assert dropped == 0
+    assert [x.title for x in kept] == ["Data Governance"]
+
+
+def test_dedup_normalized_duplicate_across_chapters():
+    # Gleicher normalisierter Titel (Case + Satzzeichen), verschiedene Kapitel,
+    # gleiche Priorität/Action -> Erstauftreten überlebt.
+    a = _concept("Data Governance", chapter="Kap1")
+    b = _concept("data governance!", chapter="Kap3")
+    kept, dropped = orch.dedup_concept_candidates([a, b])
+    assert dropped == 1
+    assert len(kept) == 1
+    assert kept[0].title == "Data Governance"  # Erstauftreten
+    assert kept[0].chapter == "Kap1"
+
+
+def test_dedup_priority_upgrade_replaces_survivor():
+    a = _concept("X", priority="medium")
+    b = _concept("X", priority="high")
+    kept, dropped = orch.dedup_concept_candidates([a, b])
+    assert dropped == 1
+    assert len(kept) == 1
+    assert kept[0].priority == "high"  # stärkere Fassung gewinnt
+
+
+def test_dedup_skip_to_actionable_upgrade():
+    a = _concept("Y", priority="medium", action="skip")
+    b = _concept("Y", priority="medium", action="create")
+    kept, dropped = orch.dedup_concept_candidates([a, b])
+    assert dropped == 1
+    assert len(kept) == 1
+    assert kept[0].action == "create"  # non-skip schlägt skip bei gleicher Prio
+
+
+def test_dedup_actionable_survivor_not_replaced_by_skip():
+    # Umkehrung: Erstauftreten actionable, zweites skip bei gleicher Prio -> bleibt.
+    a = _concept("Y", priority="medium", action="create")
+    b = _concept("Y", priority="medium", action="skip")
+    kept, dropped = orch.dedup_concept_candidates([a, b])
+    assert dropped == 1
+    assert kept[0].action == "create"
+
+
+def test_dedup_secondary_mention_untouched():
+    # Zwei secondary_mention mit gleichem Titel -> beide bleiben (eigener Kanal).
+    a = _concept("S", origin="secondary_mention")
+    b = _concept("S", origin="secondary_mention")
+    # zusätzlich ein primary mit gleichem normalisierten Titel -> kollidiert NICHT
+    p = _concept("S", origin="primary")
+    kept, dropped = orch.dedup_concept_candidates([a, p, b])
+    assert dropped == 0
+    assert len(kept) == 3
+
+
+def test_dedup_funnel_event_emitted(monkeypatch, tmp_path):
+    read = _capture(monkeypatch, tmp_path)
+    survivor = _concept("Metadaten", priority="high", chapter="Kap1")
+    dup = _concept("metadaten", priority="low", chapter="Kap2")  # schwächer -> verworfen
+
+    kept, dropped = orch.dedup_concept_candidates([survivor, dup])
+
+    assert dropped == 1
+    assert kept[0].priority == "high"
+    events = read()
+    assert len(events) == 1
+    e = events[0]
+    assert e["title"] == "metadaten"  # der verworfene Kandidat
+    assert e["stage"] == "planner_dedup"
+    assert e["outcome"] == "dropped"
+    assert e["drop_reason"] == "chapter_duplicate"
+    assert e["detail"] == "survivor=Metadaten"
+
+
+def test_dedup_empty_input():
+    kept, dropped = orch.dedup_concept_candidates([])
+    assert kept == []
+    assert dropped == 0
+
+
+# ===========================================================================
+# cap_candidates_balanced
+# ===========================================================================
+
+
+def test_cap_highs_overflow_proportional_to_word_share():
+    # 3 highs Kapitel A (700 W) + 3 highs Kapitel B (300 W), Budget 4.
+    # D'Hondt-Verteilung -> A=3, B=1.
+    concepts = [
+        _concept("A1", priority="high", chapter="A"),
+        _concept("A2", priority="high", chapter="A"),
+        _concept("A3", priority="high", chapter="A"),
+        _concept("B1", priority="high", chapter="B"),
+        _concept("B2", priority="high", chapter="B"),
+        _concept("B3", priority="high", chapter="B"),
+    ]
+    keys = ["A", "A", "A", "B", "B", "B"]
+    wc = {"A": 700, "B": 300}
+    kept, capped = orch.cap_candidates_balanced(concepts, 4, wc, keys)
+
+    assert len(kept) == 4
+    assert len(capped) == 2
+    a_kept = sum(1 for c in kept if c.chapter == "A")
+    b_kept = sum(1 for c in kept if c.chapter == "B")
+    assert a_kept == 3  # großes Kapitel bekommt mehr Slots
+    assert b_kept == 1
+    assert all(c.chapter == "B" for c in capped)
+
+
+def test_cap_rest_slots_by_word_share():
+    # Nur medium-Kandidaten -> alle Slots über die Rest-Verteilung.
+    # A (800 W) 6 medium, B (200 W) 6 medium, Budget 5 -> A=4, B=1.
+    concepts = [_concept(f"A{i}", priority="medium", chapter="A") for i in range(6)]
+    concepts += [_concept(f"B{i}", priority="medium", chapter="B") for i in range(6)]
+    keys = ["A"] * 6 + ["B"] * 6
+    wc = {"A": 800, "B": 200}
+    kept, capped = orch.cap_candidates_balanced(concepts, 5, wc, keys)
+
+    assert len(kept) == 5
+    a_kept = sum(1 for c in kept if c.chapter == "A")
+    b_kept = sum(1 for c in kept if c.chapter == "B")
+    assert a_kept > b_kept  # großes Kapitel mehr Slots
+    assert a_kept == 4
+    assert b_kept == 1
+
+
+def test_cap_priority_within_chapter():
+    # Ein Kapitel, Budget 2, Mix aus low/high/medium -> high + medium bleiben.
+    low = _concept("low", priority="low", chapter="K1")
+    high = _concept("high", priority="high", chapter="K1")
+    medium = _concept("medium", priority="medium", chapter="K1")
+    concepts = [low, high, medium]
+    keys = ["K1", "K1", "K1"]
+    wc = {"K1": 500}
+    kept, capped = orch.cap_candidates_balanced(concepts, 2, wc, keys)
+
+    kept_titles = {c.title for c in kept}
+    assert kept_titles == {"high", "medium"}
+    assert [c.title for c in capped] == ["low"]
+
+
+def test_cap_secondary_mention_budget_neutral():
+    concepts = [
+        _concept("A1", priority="high", chapter="A"),
+        _concept("A2", priority="high", chapter="A"),
+        _concept("S", origin="secondary_mention", chapter="A"),
+    ]
+    keys = ["A", "A", "A"]
+    wc = {"A": 100}
+    kept, capped = orch.cap_candidates_balanced(concepts, 1, wc, keys)
+
+    kept_titles = {c.title for c in kept}
+    # secondary_mention immer behalten, budget-neutral; nur 1 high passt ins Budget
+    assert "S" in kept_titles
+    high_kept = sum(1 for c in kept if c.origin == "primary")
+    assert high_kept == 1
+    assert len(capped) == 1
+    assert all(c.origin != "secondary_mention" for c in capped)
+
+
+def test_cap_budget_exceeds_candidate_count_is_noop():
+    concepts = [
+        _concept("A1", priority="high", chapter="A"),
+        _concept("B1", priority="medium", chapter="B"),
+    ]
+    keys = ["A", "B"]
+    wc = {"A": 100, "B": 100}
+    kept, capped = orch.cap_candidates_balanced(concepts, 100, wc, keys)
+
+    assert capped == []
+    assert [id(c) for c in kept] == [id(c) for c in concepts]  # unverändert
+
+
+def test_cap_deterministic_on_tie():
+    # Gleiche Wortzahl + gleiche Priorität -> deterministisch (frühere Reihenfolge).
+    concepts = [
+        _concept("A1", priority="medium", chapter="A"),
+        _concept("B1", priority="medium", chapter="B"),
+        _concept("A2", priority="medium", chapter="A"),
+        _concept("B2", priority="medium", chapter="B"),
+    ]
+    keys = ["A", "B", "A", "B"]
+    wc = {"A": 500, "B": 500}
+    kept1, capped1 = orch.cap_candidates_balanced(concepts, 2, wc, keys)
+    kept2, capped2 = orch.cap_candidates_balanced(concepts, 2, wc, keys)
+
+    assert [c.title for c in kept1] == [c.title for c in kept2]  # stabil
+    assert len(kept1) == 2
+    # Tie-Break: je ein Slot pro Kapitel, Erstauftreten
+    assert {c.chapter for c in kept1} == {"A", "B"}