fix(eu): drei Fehler beheben, die den Ertrag des Faktenchecks aufzehrten

Der Vergleichslauf zur Innenministerkonferenz (Lage 49) zeigte, dass der
verbesserte Faktencheck zwar zehn Fakten lieferte, davon aber nur drei
uebrig blieben. Die Ursachen lagen in drei nachgelagerten Stellen.

1. Quellenzuordnung stufte alle zehn Fakten herab.
   Durch den groesseren Belegblock stuetzt das Modell seine Fakten jetzt
   staerker auf die europaeische Suche als auf die gespeicherten Meldungen.
   Es zitiert dabei die Belegnummern statt vollstaendiger Adressen. Die
   Quellenzuordnung suchte daraufhin vergeblich nach passenden Meldungen und
   stufte jeden Fakt herab. Im Vergleich: Lauf 45 hatte null Herabstufungen,
   Lauf 44 eine, Lauf 49 zehn von zehn.
   Behoben an zwei Stellen. Die Belegsuche gibt die gefundenen Quellen jetzt
   einzeln zurueck, der Faktencheck reicht sie an die Zuordnung durch, und
   der Auftrag verlangt ausdruecklich die vollstaendige Adresse je Beleg.

2. Duplikatpruefung loeschte sieben von zehn Fakten.
   Das Clustering warf inhaltlich verschiedene Beschluesse in eine Gruppe,
   weil viele Behauptungen gleich beginnen. Es gab keinerlei Absicherung, bei
   kleinen Faktenmengen entschied das Modell sogar voellig allein.
   Zwei Sicherungen greifen jetzt. Eine Gruppe, die mehr als 40 Prozent aller
   Fakten umfasst, wird verworfen. Und jeder Loeschkandidat muss dem
   behaltenen Fakt auch rechnerisch aehnlich sein, Grenze 0,55 im selben Mass
   wie im Vorfilter. Beide Werte sind ueber Umgebungsvariablen einstellbar.

3. Leerrunden in der Recherche.
   War die Treffer-Obergrenze erreicht, brach nur die innere Schleife ab. Die
   Rundenschleife lief bis zum Ende weiter und befragte jedes Mal das teuerste
   Modell, ohne noch eine einzige Suche auszufuehren. Das kostete rund 0,17
   USD je Durchgang, bei drei Durchgaengen etwa 0,50 USD pro Aktualisierung.

Der Weg ueber die Anthropic-CLI bleibt unveraendert. Neu sind 16 Pruefungen,
darunter der nachgebildete Fall aus Lage 49, insgesamt laufen jetzt 80
Pruefungen ohne Netzzugriff und ohne Kosten.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/agents/eu_researcher.py

@@ -20,6 +20,7 @@ oder URL ableitbar ist, staan selbst liefert keine Datumsangaben.
 import asyncio
 import json
 import logging
+from dataclasses import dataclass
 from datetime import datetime
 
 from config import (
@@ -305,6 +306,17 @@ async def run_eu_research(
         if not action["queries"]:
             break
 
+        # Ist die Treffer-Obergrenze erreicht, bringen weitere Runden nichts
+        # mehr. Ohne diesen Abbruch liefe die Schleife bis zur letzten Runde
+        # weiter und befragte jedes Mal das teuerste Modell, ohne dass danach
+        # noch eine einzige Suche ausgefuehrt wuerde.
+        if len(collected) >= MAX_TOTAL_RESULTS:
+            logger.info(
+                "EU-Recherche: Suchphase nach Runde %d beendet, Treffer-Obergrenze %d erreicht",
+                round_no, MAX_TOTAL_RESULTS,
+            )
+            break
+
     total.cost_usd += searches_done * STAAN_COST_PER_QUERY_USD
 
     if not collected:
@@ -385,6 +397,9 @@ _SEARCH_NOTE_WITH_RESULTS = (
     " Nutze stattdessen die unten angehängten ERGEBNISSE DER EUROPÄISCHEN WEBSUCHE "
     "als unabhängige Zweitquellen und zitiere sie als Evidenz, wenn sie eine "
     "Behauptung stützen oder widerlegen. Zusätzlich zählen die übergebenen Meldungen."
+    " WICHTIG: Nenne bei jedem Beleg, auf den du dich stützt, immer die vollständige "
+    "Adresse (die URL hinter der [S..]-Nummer), nicht nur die Nummer. Ohne Adresse "
+    "gilt ein Beleg als nicht nachprüfbar."
 )
 _SEARCH_NOTE_NO_RESULTS = (
     " Stütze dich ausschließlich auf die übergebenen Meldungen und Quellen."
@@ -432,6 +447,21 @@ def _add_usage(total: ClaudeUsage, u: ClaudeUsage | None) -> None:
     total.duration_ms += u.duration_ms
 
 
+@dataclass
+class Belegsuche:
+    """Ergebnis einer gezielten Belegsuche.
+
+    block    = fertiger Kontextblock fuer den Auftrag, leer wenn nichts gefunden
+    usage    = Verbrauch aus Planung und Suche
+    queries  = tatsaechlich ausgefuehrte Suchanfragen, fuer die Nachrunde
+    quellen  = gefundene Quellen einzeln, fuer die Quellenzuordnung im Faktencheck
+    """
+    block: str
+    usage: ClaudeUsage
+    queries: list[str]
+    quellen: list[dict]
+
+
 async def plan_verification_searches(
     *,
     title: str,
@@ -440,22 +470,23 @@ async def plan_verification_searches(
     max_queries: int = EU_VERIFY_MAX_QUERIES,
     avoid_queries: list[str] | None = None,
     label: str = "Stützsuche",
-) -> tuple[str, ClaudeUsage, list[str]]:
+) -> Belegsuche:
     """Plant gezielte Belegsuchen zu den uebergebenen Punkten und fuehrt sie aus.
 
     Die Planung erfolgt punktbezogen, das planende Modell ordnet jeder Anfrage
     zu, welche Punkte sie belegen soll. avoid_queries verhindert, dass eine
     Nachrunde dieselben Anfragen noch einmal stellt.
 
-    Gibt (kontextblock, verbrauch, ausgefuehrte_anfragen) zurueck. Der
+    Gibt eine Belegsuche zurueck. Der Kontextblock ist leer, wenn nichts
-    Kontextblock ist leer, wenn nichts gefunden wurde.
+    gefunden wurde. Die gefundenen Quellen werden zusaetzlich einzeln
+    zurueckgegeben, damit die Quellenzuordnung im Faktencheck sie kennt.
     """
     from config import CLAUDE_MODEL_FAST
 
     total = ClaudeUsage()
     items = [str(s).strip() for s in (search_items or []) if str(s).strip()][:EU_VERIFY_MAX_ITEMS]
     if not items:
-        return "", total, []
+        return Belegsuche("", total, [], [])
 
     default_market = market_for_language(_iso_from_display(output_language))
     avoid_block = ""
@@ -479,7 +510,7 @@ async def plan_verification_searches(
         _add_usage(total, plan_usage)
     except Exception as e:
         logger.warning("EU-%s: Planung der Anfragen fehlgeschlagen (%s), fahre ohne Suche fort", label, e)
-        return "", total, []
+        return Belegsuche("", total, [], [])
 
     from agents.researcher import _extract_json_object
     obj = _extract_json_object(plan_text or "")
@@ -498,10 +529,11 @@ async def plan_verification_searches(
 
     if not queries:
         logger.warning("EU-%s: Planung lieferte keine brauchbaren Anfragen", label)
-        return "", total, []
+        return Belegsuche("", total, [], [])
 
     lines: list[str] = []
     ausgefuehrt: list[str] = []
+    quellen: list[dict] = []
     gesehene_urls: set[str] = set()
     treffer = 0
     for q in queries:
@@ -524,6 +556,15 @@ async def plan_verification_searches(
             if url:
                 gesehene_urls.add(url)
             treffer += 1
+            # Die Quelle wird zusaetzlich einzeln gefuehrt, damit die
+            # Quellenzuordnung im Faktencheck sie wie eine Meldung behandeln
+            # kann. Ohne das gelten Fakten als unbelegt, die sich allein auf
+            # diese Belege stuetzen.
+            quellen.append({
+                "headline": r["title"] or "",
+                "source": r["hostname"] or "",
+                "source_url": url,
+            })
             lines.append(f"[S{treffer}] {r['title'] or '(ohne Titel)'} | {r['hostname']} | {r['url']}")
             if r.get("snippet"):
                 lines.append(f"     Auszug: {r['snippet'][:300]}")
@@ -534,14 +575,14 @@ async def plan_verification_searches(
     logger.info("EU-%s: %d Suchen, %d Treffer im Kontextblock", label, len(ausgefuehrt), treffer)
 
     if not lines:
-        return "", total, ausgefuehrt
+        return Belegsuche("", total, ausgefuehrt, [])
 
     block = (
         "\n\nERGEBNISSE DER EUROPÄISCHEN WEBSUCHE "
         "(von unserem System ausgeführt, als unabhängige Zweitquellen nutzbar):\n"
         + "\n".join(lines)
     )
-    return block, total, ausgefuehrt
+    return Belegsuche(block, total, ausgefuehrt, quellen)
 
 
 def build_eu_prompt(prompt: str, results_block: str = "") -> str:
@@ -571,11 +612,12 @@ async def eu_call_with_search(
     total = ClaudeUsage()
     results_block = ""
     if search_items:
-        results_block, such_usage, _ = await plan_verification_searches(
+        suche = await plan_verification_searches(
             title=title, search_items=search_items,
             output_language=output_language, max_queries=max_queries,
         )
-        _add_usage(total, such_usage)
+        results_block = suche.block
+        _add_usage(total, suche.usage)
 
     result, usage = await call_claude(build_eu_prompt(prompt, results_block), tools=None)
     _add_usage(total, usage)

src/agents/factchecker.py

@@ -551,15 +551,18 @@ class FactCheckerAgent:
         anzahl_anfragen = EU_VERIFY_MAX_QUERIES if max_queries is None else max_queries
         anzahl_nachhak = EU_VERIFY_FOLLOWUP_QUERIES if followup_queries is None else followup_queries
 
-        block, such_usage, gestellte_anfragen = await plan_verification_searches(
+        suche = await plan_verification_searches(
             title=title, search_items=search_items,
             output_language=output_language, max_queries=anzahl_anfragen,
         )
-        _add_usage(gesamt, such_usage)
+        _add_usage(gesamt, suche.usage)
 
-        result, usage = await call_claude(build_eu_prompt(prompt, block), tools=None)
+        result, usage = await call_claude(build_eu_prompt(prompt, suche.block), tools=None)
         _add_usage(gesamt, usage)
-        facts = self._parse_response(result, articles=articles)
+        # Die Suchbelege zaehlen bei der Quellenzuordnung wie Meldungen. Ohne
+        # sie gelten Fakten als unbelegt, die sich allein auf die europaeische
+        # Suche stuetzen, und werden faelschlich herabgestuft.
+        facts = self._parse_response(result, articles=(articles or []) + suche.quellen)
 
         offene = [f for f in facts if f.get("status") in OFFENE_STATUS]
         if anzahl_nachhak <= 0 or len(offene) < EU_VERIFY_FOLLOWUP_MIN_OPEN:
@@ -569,7 +572,7 @@ class FactCheckerAgent:
         facts, nachhak_usage = await self._eu_nachhaken(
             facts=facts, offene=offene, title=title,
             output_language=output_language, incident_type=incident_type,
-            avoid_queries=gestellte_anfragen, followup_queries=anzahl_nachhak,
+            avoid_queries=suche.queries, followup_queries=anzahl_nachhak,
         )
         _add_usage(gesamt, nachhak_usage)
         return facts, gesamt
@@ -600,12 +603,13 @@ class FactCheckerAgent:
         if not claims:
             return facts, gesamt
 
-        block, such_usage, _ = await plan_verification_searches(
+        suche = await plan_verification_searches(
             title=title, search_items=claims, output_language=output_language,
             max_queries=(EU_VERIFY_FOLLOWUP_QUERIES if followup_queries is None else followup_queries),
             avoid_queries=avoid_queries, label="Nachhak-Suche",
         )
-        _add_usage(gesamt, such_usage)
+        block = suche.block
+        _add_usage(gesamt, suche.usage)
         if not block:
             logger.info("EU-Nachhak-Runde: keine zusaetzlichen Belege gefunden, Fakten bleiben unveraendert")
             return facts, gesamt

src/services/post_refresh_qc.py

@@ -25,6 +25,34 @@ STATUS_PRIORITY = {
     "developing": 1,
 }
 
+# Schutz gegen Fehlurteile beim Zusammenfassen von Duplikaten. Das Clustering
+# entscheidet ein schnelles Modell, das gelegentlich inhaltlich verschiedene
+# Fakten in eine Gruppe wirft, etwa wenn viele Behauptungen gleich beginnen.
+# Zwei Sicherungen greifen. Eine Gruppe, die einen zu grossen Anteil aller
+# Fakten umfasst, wird komplett verworfen. Und jeder einzelne Loeschkandidat
+# muss dem behaltenen Fakt auch rechnerisch aehnlich genug sein.
+DEDUP_MAX_CLUSTER_ANTEIL = float(os.environ.get("QC_DEDUP_MAX_CLUSTER_ANTEIL", "0.4"))
+DEDUP_MIN_AEHNLICHKEIT = float(os.environ.get("QC_DEDUP_MIN_AEHNLICHKEIT", "0.55"))
+
+
+def _aehnlichkeit(claim_a: str, claim_b: str) -> float:
+    """Rechnerische Aehnlichkeit zweier Behauptungen zwischen 0 und 1.
+
+    Gleiche Gewichtung wie im Vorfilter, damit beide Stufen dasselbe Mass
+    verwenden.
+    """
+    from agents.factchecker import normalize_claim, _keyword_set
+
+    norm_a = normalize_claim(claim_a or "")
+    norm_b = normalize_claim(claim_b or "")
+    if not norm_a or not norm_b:
+        return 0.0
+    kw_a = _keyword_set(claim_a or "")
+    kw_b = _keyword_set(claim_b or "")
+    kw_union = kw_a | kw_b
+    jaccard = len(kw_a & kw_b) / len(kw_union) if kw_union else 0.0
+    return 0.7 * SequenceMatcher(None, norm_a, norm_b).ratio() + 0.3 * jaccard
+
 # ---------------------------------------------------------------------------
 # 1. Faktencheck-Duplikate
 # ---------------------------------------------------------------------------
@@ -177,11 +205,23 @@ async def check_fact_duplicates(db, incident_id: int, incident_title: str) -> in
     facts_by_id = {f["id"]: f for f in all_facts}
     ids_to_delete = set()
 
+    max_cluster = max(2, int(len(all_facts) * DEDUP_MAX_CLUSTER_ANTEIL))
+
     for cluster_ids in all_clusters:
         valid_ids = [cid for cid in cluster_ids if cid in facts_by_id]
         if len(valid_ids) <= 1:
             continue
 
+        # Sicherung 1: Eine Gruppe, die einen zu grossen Teil aller Fakten
+        # umfasst, ist fast immer ein Fehlurteil des Clusterings.
+        if len(valid_ids) > max_cluster:
+            logger.warning(
+                "QC Duplikat: Gruppe mit %d von %d Fakten verworfen, das ist "
+                "unplausibel (Grenze %d)",
+                len(valid_ids), len(all_facts), max_cluster,
+            )
+            continue
+
         cluster_facts = [facts_by_id[cid] for cid in valid_ids]
         best = max(cluster_facts, key=lambda f: (
             STATUS_PRIORITY.get(f["status"], 0),
@@ -190,12 +230,23 @@ async def check_fact_duplicates(db, incident_id: int, incident_title: str) -> in
         ))
 
         for fact in cluster_facts:
-            if fact["id"] != best["id"]:
+            if fact["id"] == best["id"]:
-                ids_to_delete.add(fact["id"])
+                continue
-                logger.info(
+            # Sicherung 2: Nur loeschen, wenn die beiden Behauptungen auch
-                    "QC Duplikat: ID %d entfernt, behalte ID %d ('%s')",
+            # rechnerisch nah beieinander liegen.
-                    fact["id"], best["id"], best["claim"][:60],
+            naehe = _aehnlichkeit(fact["claim"], best["claim"])
+            if naehe < DEDUP_MIN_AEHNLICHKEIT:
+                logger.warning(
+                    "QC Duplikat: ID %d behalten, Aehnlichkeit zu ID %d nur %.2f "
+                    "('%s')",
+                    fact["id"], best["id"], naehe, fact["claim"][:60],
                 )
+                continue
+            ids_to_delete.add(fact["id"])
+            logger.info(
+                "QC Duplikat: ID %d entfernt (Aehnlichkeit %.2f), behalte ID %d ('%s')",
+                fact["id"], naehe, best["id"], best["claim"][:60],
+            )
 
     if ids_to_delete:
         placeholders = ",".join("?" * len(ids_to_delete))

tests/test_eu_belegsuche.py

@@ -80,10 +80,11 @@ plan_antwort["queries"] = [
 treffer_je_anfrage["Ceuta Opferzahl offiziell"] = treffer(2, "a")
 treffer_je_anfrage["Ceuta border deaths toll"] = treffer(2, "b")
 treffer_je_anfrage["Sanchez Ceuta declaracion"] = treffer(2, "c")
-block, usage, ausgefuehrt = asyncio.run(eur.plan_verification_searches(
+_r = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["57 Tote gemeldet", "Sanchez sprach von Angriff"],
     output_language="Deutsch", max_queries=7,
 ))
+block, usage, ausgefuehrt = _r.block, _r.usage, _r.queries
 pruefe("A1 alle geplanten Anfragen ausgefuehrt", len(zustand["suchen"]) == 3, zustand["suchen"])
 pruefe("A2 ausgefuehrte Anfragen zurueckgemeldet", len(ausgefuehrt) == 3, ausgefuehrt)
 pruefe("A3 jeder Punkt steht im Planungsauftrag",
@@ -106,9 +107,10 @@ plan_antwort["queries"] = [
 ]
 treffer_je_anfrage["anfrage eins"] = treffer(3, "x")
 treffer_je_anfrage["anfrage zwei"] = treffer(3, "x")  # identische URLs
-block_b, _, _ = asyncio.run(eur.plan_verification_searches(
+_rb = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["Punkt"], max_queries=7,
 ))
+block_b = _rb.block
 pruefe("B1 Dubletten entfernt", block_b.count("[S") == 3, block_b.count("[S"))
 pruefe("B2 Nummerierung bleibt luecklos",
        all(("[S" + str(i) + "]") in block_b for i in (1, 2, 3)))
@@ -120,10 +122,11 @@ plan_antwort["queries"] = [
     {"q": "neue anfrage", "market": "de-de"},
 ]
 treffer_je_anfrage["neue anfrage"] = treffer(2, "n")
-block_c, _, ausgefuehrt_c = asyncio.run(eur.plan_verification_searches(
+_rc = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["Punkt"], max_queries=5,
     avoid_queries=["alte anfrage"], label="Nachhak-Suche",
 ))
+block_c, ausgefuehrt_c = _rc.block, _rc.queries
 pruefe("C1 bereits gestellte Anfrage wird verworfen",
        [s["q"] for s in zustand["suchen"]] == ["neue anfrage"], zustand["suchen"])
 pruefe("C2 Sperrliste steht im Planungsauftrag", "alte anfrage" in zustand["plan_prompt"])
@@ -134,9 +137,10 @@ neu()
 plan_antwort["queries"] = [{"q": "q" + str(i), "market": "de-de"} for i in range(1, 11)]
 for i in range(1, 11):
     treffer_je_anfrage["q" + str(i)] = treffer(20, "y" + str(i))
-block_d, _, ausgefuehrt_d = asyncio.run(eur.plan_verification_searches(
+_rd = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["Punkt"], max_queries=4,
 ))
+block_d, ausgefuehrt_d = _rd.block, _rd.queries
 pruefe("D1 hoechstens max_queries Anfragen", len(ausgefuehrt_d) == 4, ausgefuehrt_d)
 pruefe("D2 Treffer je Anfrage begrenzt",
        block_d.count("[S") == 4 * EU_VERIFY_HITS_PER_QUERY, block_d.count("[S"))
@@ -163,10 +167,11 @@ async def staan_mit_fehler(q, **kw):
 
 
 eur.staan_search = staan_mit_fehler
-block_e, usage_e, ausgefuehrt_e = asyncio.run(eur.plan_verification_searches(
+_re = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["Punkt"], max_queries=5,
 ))
 eur.staan_search = fake_staan
+block_e, usage_e, ausgefuehrt_e = _re.block, _re.usage, _re.queries
 pruefe("E1 gescheiterte Suche uebersprungen", ausgefuehrt_e == ["heil"], ausgefuehrt_e)
 pruefe("E2 gefundene Belege bleiben erhalten", block_e.count("[S") == 2, block_e.count("[S"))
 pruefe("E3 nur bezahlte Suchen berechnet",
@@ -174,15 +179,48 @@ pruefe("E3 nur bezahlte Suchen berechnet",
 
 neu()
 plan_antwort["queries"] = []
-block_f, _, ausgefuehrt_f = asyncio.run(eur.plan_verification_searches(
+_rf = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=["Punkt"], max_queries=5,
 ))
+block_f, ausgefuehrt_f = _rf.block, _rf.queries
 pruefe("E4 leere Planung liefert leeren Block", block_f == "" and ausgefuehrt_f == [])
 
-block_g, _, _ = asyncio.run(eur.plan_verification_searches(
+_rg = asyncio.run(eur.plan_verification_searches(
     title="Ceuta", search_items=[], max_queries=5,
 ))
+block_g = _rg.block
 pruefe("E5 ohne Pruefpunkte keine Suche", block_g == "")
 
+print("\nF) Gefundene Quellen werden einzeln zurueckgegeben")
+neu()
+plan_antwort["queries"] = [{"q": "eine anfrage", "market": "de-de"}]
+treffer_je_anfrage["eine anfrage"] = treffer(3, "q", host="tagesschau.de")
+_rh = asyncio.run(eur.plan_verification_searches(
+    title="Ceuta", search_items=["Punkt"], max_queries=5,
+))
+pruefe("F1 jede Quelle einzeln gemeldet", len(_rh.quellen) == 3, len(_rh.quellen))
+pruefe("F2 Quelle traegt Titel, Herkunft und Adresse",
+       all(q.get("headline") and q.get("source") and q.get("source_url") for q in _rh.quellen),
+       _rh.quellen[:1])
+pruefe("F3 Adressen stimmen mit dem Block ueberein",
+       all(q["source_url"] in _rh.block for q in _rh.quellen))
+neu()
+plan_antwort["queries"] = [{"q": "a", "market": "de-de"}, {"q": "b", "market": "de-de"}]
+treffer_je_anfrage["a"] = treffer(2, "dup")
+treffer_je_anfrage["b"] = treffer(2, "dup")
+_ri = asyncio.run(eur.plan_verification_searches(
+    title="Ceuta", search_items=["Punkt"], max_queries=5,
+))
+pruefe("F4 doppelte Quellen nur einmal gemeldet", len(_ri.quellen) == 2, len(_ri.quellen))
+
+print("\nG) Der Auftrag verlangt vollstaendige Quellenadressen")
+from agents.eu_researcher import build_eu_prompt  # noqa: E402
+
+mit = build_eu_prompt("Auftrag", "\n\nBELEGE\n[S1] Titel | host | https://x.y/z")
+ohne = build_eu_prompt("Auftrag", "")
+pruefe("G1 Hinweis auf die Adresspflicht vorhanden",
+       "vollständige" in mit and "URL" in mit)
+pruefe("G2 ohne Belege kein Adresshinweis", "Adresse (die URL" not in ohne)
+
 print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
 sys.exit(1 if fail else 0)

tests/test_eu_factcheck.py

@@ -35,6 +35,13 @@ def pruefe(name, bedingung, extra=""):
 aufrufe = {"such": [], "modell": []}
 
 
+SUCHQUELLE = {
+    "headline": "Innenministerkonferenz beschliesst Fahrplan",
+    "source": "example.org",
+    "source_url": "https://example.org/a",
+}
+
+
 async def fake_plan(*, title, search_items, output_language="Deutsch",
                     max_queries=7, avoid_queries=None, label="Stuetzsuche"):
     aufrufe["such"].append({
@@ -45,7 +52,7 @@ async def fake_plan(*, title, search_items, output_language="Deutsch",
     })
     u = ClaudeUsage(input_tokens=10, output_tokens=5, cost_usd=0.01)
     block = "\n\nBELEGE [" + label + "]\n[S1] Beleg | example.org | https://example.org/a"
-    return block, u, [label + "-q1", label + "-q2"]
+    return eur.Belegsuche(block, u, [label + "-q1", label + "-q2"], [dict(SUCHQUELLE)])
 
 
 antworten = []
@@ -240,5 +247,33 @@ asyncio.run(fc._eu_pruefen(
 ))
 pruefe("I1 followup_queries=0 unterbindet die Nachrunde", len(aufrufe["such"]) == 1, len(aufrufe["such"]))
 
+print("\nJ) Suchbelege zaehlen bei der Quellenzuordnung")
+_ai_backend_var.set("bedrock")
+neu()
+# Behauptung, die zu KEINEM gespeicherten Artikel passt, wohl aber zur
+# Suchquelle. Vor der Korrektur wurde so ein Fakt herabgestuft.
+NUR_SUCHE = "Die Innenministerkonferenz beschliesst einen Fahrplan"
+antworten.append(json_fakten([
+    (NUR_SUCHE, "established", "belegt durch die Suche"),
+    (F_GRENZE, "confirmed", "belegt"),
+]))
+facts_j, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch"))
+treffer_j = [f for f in facts_j if f["claim"].startswith("Die Innenministerkonferenz")]
+pruefe("J1 Fakt allein aus der Suche bleibt belegt",
+       treffer_j and treffer_j[0]["status"] == "established",
+       [(f["claim"][:35], f["status"]) for f in facts_j])
+pruefe("J2 Adresse der Suchquelle steht in der Evidenz",
+       treffer_j and "https://example.org/a" in treffer_j[0]["evidence"],
+       treffer_j[0]["evidence"][:160] if treffer_j else "")
+
+print("\nK) Ohne Suchbelege bleibt die Zuordnung wie bisher")
+_ai_backend_var.set("cli")
+neu()
+antworten.append(json_fakten([(NUR_SUCHE, "established", "belegt")]))
+facts_k, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch"))
+pruefe("K1 CLI stuft ohne zuordenbare Quelle weiterhin herab",
+       facts_k and facts_k[0]["status"] != "established",
+       [(f["claim"][:35], f["status"]) for f in facts_k])
+
 print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
 sys.exit(1 if fail else 0)

tests/test_qc_und_runden.py

@@ -0,0 +1,213 @@
+"""Testet die Absicherung der Duplikatpruefung und den Abbruch der Suchrunden.
+
+Laeuft ohne Netzzugriff und ohne Kosten, Datenbank, Modell und Suche sind
+durch Testdoubles ersetzt.
+
+Aufruf aus dem Projektstamm:
+    venv/bin/python tests/test_qc_und_runden.py
+"""
+import asyncio
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src"))
+
+import services.post_refresh_qc as qc  # noqa: E402
+import agents.eu_researcher as eur  # noqa: E402
+from agents.claude_client import ClaudeUsage  # noqa: E402
+
+ok = 0
+fail = 0
+
+
+def pruefe(name, bedingung, extra=""):
+    global ok, fail
+    if bedingung:
+        ok += 1
+        print("  OK    " + name)
+    else:
+        fail += 1
+        print("  FEHL  " + name + " " + str(extra))
+
+
+# ---------------------------------------------------------------------------
+# Testdouble fuer die Datenbank
+# ---------------------------------------------------------------------------
+class FakeCursor:
+    def __init__(self, rows):
+        self._rows = rows
+
+    async def fetchall(self):
+        return self._rows
+
+    async def fetchone(self):
+        return self._rows[0] if self._rows else None
+
+
+class FakeDB:
+    """Liefert eine feste Faktenliste und merkt sich die Loeschauftraege."""
+
+    def __init__(self, fakten):
+        self.fakten = fakten
+        self.geloescht = []
+
+    async def execute(self, sql, params=()):
+        if sql.strip().upper().startswith("DELETE"):
+            self.geloescht.extend(params)
+            return FakeCursor([])
+        return FakeCursor(list(self.fakten))
+
+
+def fakt(fid, claim, status="established", quellen=3):
+    return {"id": fid, "claim": claim, "status": status, "sources_count": quellen,
+            "evidence": "e", "checked_at": "2026-08-01 10:00:00"}
+
+
+FAHRPLAN = "Die IMK beschloss einen gemeinsamen Bund-Laender-Fahrplan zum Aufbau der zivilen Verteidigungsfaehigkeit bis 2029"
+FAHRPLAN_ANDERS = "Die IMK beschloss einen Bund-Laender-Fahrplan fuer die zivile Verteidigung bis zum Jahr 2029"
+BUNDESWEHR = "Die IMK beschloss die institutionelle Einbindung der Bundeswehr in die Innenministerkonferenz"
+SOZIAL = "Die IMK beschloss Massnahmen zur haerteren Bekaempfung von organisiertem Sozialmissbrauch"
+MILLIARDEN = "Die IMK forderte zusaetzliche Milliardeninvestitionen des Bundes in den Zivilschutz"
+ABSCHIEBUNG = "Die Innenminister waren sich einig, dass der Bund bei Abschiebungen mehr leisten muss"
+GDP = "Die Gewerkschaft der Polizei bewertete die Beschluesse als grundsaetzlich richtig"
+SYRER = "In der Frage der Rueckfuehrung von Syrern besteht ein Konflikt zwischen SPD und CDU"
+IMK_DATUM = "Die 225. Innenministerkonferenz fand im Juni 2026 in Hamburg statt"
+BESCHLUESSE = "Die IMK fasste insgesamt 66 Beschluesse"
+
+geplante_cluster = []
+
+
+async def fake_cluster(facts, incident_title):
+    return geplante_cluster
+
+
+qc._haiku_find_duplicate_clusters = fake_cluster
+
+
+def lauf_qc(fakten, cluster):
+    global geplante_cluster
+    geplante_cluster = cluster
+    db = FakeDB(fakten)
+    entfernt = asyncio.run(qc.check_fact_duplicates(db, 49, "Innenministerkonferenz 2026"))
+    return entfernt, db.geloescht
+
+
+print("\nA) Der Fall aus Lage 49, sieben verschiedene Fakten in einer Gruppe")
+zehn = [
+    fakt(469, FAHRPLAN), fakt(468, IMK_DATUM), fakt(470, BUNDESWEHR),
+    fakt(471, SOZIAL), fakt(472, MILLIARDEN), fakt(473, ABSCHIEBUNG),
+    fakt(474, SYRER, "disputed"), fakt(475, BESCHLUESSE), fakt(476, GDP),
+    fakt(477, "Die IMK befasste sich mit der Umsetzung des europaeischen Asylsystems"),
+]
+entfernt, geloescht = lauf_qc(zehn, [[468, 469, 470, 471, 472, 473, 475, 476]])
+pruefe("A1 unplausible Gruppe wird komplett verworfen", entfernt == 0, entfernt)
+pruefe("A2 kein Fakt geloescht", geloescht == [], geloescht)
+
+print("\nB) Echte Dubletten werden weiterhin entfernt")
+entfernt, geloescht = lauf_qc(
+    [fakt(1, FAHRPLAN, quellen=8), fakt(2, FAHRPLAN_ANDERS, quellen=2),
+     fakt(3, BUNDESWEHR), fakt(4, SOZIAL), fakt(5, MILLIARDEN)],
+    [[1, 2]])
+pruefe("B1 echtes Duplikat entfernt", entfernt == 1, entfernt)
+pruefe("B2 der besser belegte Fakt bleibt", geloescht == [2], geloescht)
+
+print("\nC) Kleine Gruppe mit unaehnlichen Fakten wird nicht geloescht")
+entfernt, geloescht = lauf_qc(
+    [fakt(1, FAHRPLAN), fakt(2, SOZIAL), fakt(3, BUNDESWEHR), fakt(4, MILLIARDEN),
+     fakt(5, ABSCHIEBUNG), fakt(6, GDP)],
+    [[1, 2]])
+pruefe("C1 unaehnlicher Fakt bleibt erhalten", entfernt == 0, entfernt)
+
+print("\nD) Gemischte Gruppe, nur der wirklich aehnliche Fakt faellt weg")
+entfernt, geloescht = lauf_qc(
+    [fakt(1, FAHRPLAN, quellen=9), fakt(2, FAHRPLAN_ANDERS, quellen=1), fakt(3, SOZIAL),
+     fakt(4, BUNDESWEHR), fakt(5, MILLIARDEN), fakt(6, ABSCHIEBUNG),
+     fakt(7, GDP), fakt(8, SYRER), fakt(9, BESCHLUESSE), fakt(10, IMK_DATUM)],
+    [[1, 2, 3]])
+pruefe("D1 genau ein Fakt entfernt", entfernt == 1, entfernt)
+pruefe("D2 der aehnliche wurde entfernt", geloescht == [2], geloescht)
+
+print("\nE) Grenzwerte")
+pruefe("E1 Anteilsgrenze gesetzt", 0 < qc.DEDUP_MAX_CLUSTER_ANTEIL <= 1, qc.DEDUP_MAX_CLUSTER_ANTEIL)
+pruefe("E2 Aehnlichkeitsgrenze gesetzt", 0 < qc.DEDUP_MIN_AEHNLICHKEIT < 1, qc.DEDUP_MIN_AEHNLICHKEIT)
+pruefe("E3 verschiedene Sachverhalte liegen unter der Grenze",
+       qc._aehnlichkeit(FAHRPLAN, SOZIAL) < qc.DEDUP_MIN_AEHNLICHKEIT,
+       round(qc._aehnlichkeit(FAHRPLAN, SOZIAL), 2))
+pruefe("E4 echte Dublette liegt ueber der Grenze",
+       qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS) >= qc.DEDUP_MIN_AEHNLICHKEIT,
+       round(qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS), 2))
+pruefe("E5 bei zwei Fakten bleibt die Gruppengrenze nutzbar",
+       max(2, int(2 * qc.DEDUP_MAX_CLUSTER_ANTEIL)) == 2)
+
+# ---------------------------------------------------------------------------
+# Suchrunden
+# ---------------------------------------------------------------------------
+print("\nF) Die Suchphase endet, sobald die Treffergrenze erreicht ist")
+
+runden = {"n": 0}
+
+
+async def fake_bedrock_research(prompt, model=None, raw_text=False, timeout=None):
+    runden["n"] += 1
+    # Jede Runde fordert vier Suchen an, das Modell will nie von selbst enden.
+    antwort = {"action": "search", "reason": "weiter",
+               "queries": [{"q": f"anfrage {runden['n']}-{i}", "market": "de-de",
+                            "full_content": False} for i in range(1, 5)]}
+    return json.dumps(antwort), ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05)
+
+
+async def fake_staan_research(q, market="de-de", extra_snippets=True, max_snippets=4,
+                              full_content=False, extra_exclude=None, timeout=None):
+    # Jede Suche liefert 20 neue Treffer, nach zwei Runden ist die Grenze voll.
+    basis = q.replace(" ", "")
+    return [{
+        "title": "Treffer " + basis + str(i),
+        "hostname": "example.org",
+        "url": "https://example.org/" + basis + "/" + str(i),
+        "snippet": "Auszug",
+        "extra_snippets": [],
+        "full_text": "",
+    } for i in range(1, 21)]
+
+
+letzter_prompt = {"text": ""}
+
+
+async def fake_bedrock_final(prompt, model=None, raw_text=False, timeout=None):
+    """Ab dem Abschlussauftrag wird eine Auswahl zurueckgegeben."""
+    letzter_prompt["text"] = prompt
+    return "[]", ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05)
+
+
+async def bedrock_weiche(prompt, model=None, raw_text=False, timeout=None):
+    # Der Rundenauftrag traegt die Rundennummer im Kopf, der Abschlussauftrag
+    # nicht. Daran laesst sich beides sicher unterscheiden.
+    if "Es ist Runde " in prompt:
+        return await fake_bedrock_research(prompt, model, raw_text, timeout)
+    return await fake_bedrock_final(prompt, model, raw_text, timeout)
+
+
+eur.call_bedrock = bedrock_weiche
+eur.staan_search = fake_staan_research
+
+runden["n"] = 0
+text, usage = asyncio.run(eur.run_eu_research(
+    title="Innenministerkonferenz 2026", description="", incident_type="research",
+    lang_instruction="", existing_context="", preferred_sources_block="",
+    output_language="Deutsch", excluded_sources=[], research_language_iso="de",
+))
+# Die erste Runde fuellt die Grenze von 60 Treffern, danach ist Schluss.
+# Ohne die Korrektur liefe die Schleife alle fuenf Runden durch und befragte
+# jedes Mal das teuerste Modell, ohne noch zu suchen.
+suchrunden = runden["n"]
+pruefe("F1 genau eine Suchrunde statt fuenf", suchrunden == 1, suchrunden)
+pruefe("F2 vier Leerrunden gespart", eur.EU_RESEARCH_MAX_ROUNDS_RESEARCH - suchrunden == 4,
+       eur.EU_RESEARCH_MAX_ROUNDS_RESEARCH - suchrunden)
+pruefe("F3 Recherche liefert trotzdem ein Ergebnis", text == "[]", text[:40])
+pruefe("F4 Kosten nur fuer die tatsaechlich gelaufene Runde plus Abschluss",
+       round(usage.cost_usd, 2) == round(0.05 * 2 + 4 * eur.STAAN_COST_PER_QUERY_USD, 2),
+       round(usage.cost_usd, 4))
+
+print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
+sys.exit(1 if fail else 0)