AegisSight-Monitor/src/agents/orchestrator.py

"""Agenten-Orchestrierung: Queue und Steuerung der Claude-Agenten."""
import asyncio
import json
import logging
import os
import re
from datetime import datetime
from config import TIMEZONE
from typing import Optional
from urllib.parse import urlparse, urlunparse, quote_plus

import httpx

from agents.claude_client import UsageAccumulator, _cancel_event_var, _ai_backend_var
from agents import bedrock_client
from agents.factchecker import find_matching_claim, deduplicate_new_facts, TWOPHASE_MIN_FACTS
from source_rules import (
    _detect_category,
    _extract_domain,
    discover_source,
    domain_to_display_name,
)

logger = logging.getLogger("osint.orchestrator")

# Reputations-Score nach Quellenkategorie (fuer Relevanz-Scoring).
# Keys muessen mit den tatsaechlichen DB-Werten in sources.category uebereinstimmen
# (siehe DOMAIN_CATEGORY_MAP in source_rules.py).
CATEGORY_REPUTATION = {
    "nachrichtenagentur":    1.0,   # Reuters, AP, dpa, AFP — Primärquellen
    "behoerde":              1.0,   # BMI, BSI, Europol — offizielle Quellen
    "oeffentlich-rechtlich": 0.95,  # tagesschau, ZDF, ARD, BBC, ORF
    "qualitaetszeitung":     0.85,  # Spiegel, Zeit, FAZ, NZZ, Süddeutsche
    "think-tank":            0.85,  # SWP, IISS, Brookings, Chatham House
    "fachmedien":            0.8,   # heise, golem, netzpolitik, Handelsblatt
    "international":         0.75,  # CNN, Guardian, NYT, Al Jazeera, France24
    "regional":              0.65,  # regionale Tageszeitungen
    "telegram":              0.5,   # OSINT-Kanaele — gemischte Qualitaet
    "x":                     0.4,   # X/Twitter-Accounts, hohes Rauschen
    "sonstige":              0.4,   # unkategorisiert
    "boulevard":             0.3,   # Bild, Sun etc.
}

# Research-Modus: Automatisch 3 Durchläufe für optimale Ergebnisse
RESEARCH_MULTI_PASS_COUNT = 3
RESEARCH_PASS_LABELS = {1: "Breite Erfassung", 2: "Vertiefung", 3: "Konsolidierung"}


def _normalize_url(url: str) -> str:
    """URL normalisieren für Duplikat-Erkennung."""
    if not url:
        return ""
    url = url.strip()
    try:
        parsed = urlparse(url)
        # Scheme normalisieren
        scheme = parsed.scheme.lower() or "https"
        # Host normalisieren (www entfernen, lowercase)
        netloc = parsed.netloc.lower()
        if netloc.startswith("www."):
            netloc = netloc[4:]
        # Pfad normalisieren (trailing slash entfernen)
        path = parsed.path.rstrip("/")
        # Query-Parameter und Fragment entfernen (Tracking-Params etc.)
        return urlunparse((scheme, netloc, path, "", "", ""))
    except Exception:
        return url.lower().strip().rstrip("/")


def _normalize_headline(headline: str) -> str:
    """Überschrift normalisieren für Ähnlichkeitsvergleich."""
    if not headline:
        return ""
    h = headline.lower().strip()
    # Umlaute normalisieren
    h = h.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss")
    # Sonderzeichen entfernen
    h = re.sub(r"[^\w\s]", "", h)
    h = re.sub(r"\s+", " ", h).strip()
    return h


def _is_duplicate(article: dict, seen_urls: set, seen_headlines: set) -> bool:
    """Prüft ob ein Artikel ein Duplikat ist (URL oder Headline)."""
    url = article.get("source_url", "")
    headline = article.get("headline", "")

    # URL-Duplikat
    if url:
        norm_url = _normalize_url(url)
        if norm_url in seen_urls:
            return True
        seen_urls.add(norm_url)

    # Headline-Duplikat (nur wenn Überschrift lang genug)
    if headline and len(headline) > 20:
        norm_headline = _normalize_headline(headline)
        if norm_headline and norm_headline in seen_headlines:
            return True
        if norm_headline:
            seen_headlines.add(norm_headline)

    return False


def _score_relevance(article: dict, search_words: list[str] = None) -> float:
    """Berechnet einen Relevanz-Score (0.0-1.0) für einen Artikel.

    Gewichtung:
    - 40% Keyword-Dichte (wie gut passt der Artikel zum Suchbegriff)
    - 30% Quellen-Reputation (basierend auf Kategorie)
    - 20% Inhaltstiefe (hat der Artikel substantiellen Inhalt)
    - 10% RSS-Score (falls vorhanden, vom RSS-Parser)
    """
    score = 0.0

    # 1. Keyword-Dichte (40%)
    rss_score = article.get("relevance_score", 0.0)
    if rss_score > 0:
        score += 0.4 * rss_score
    elif search_words:
        text = f"{article.get('headline', '')} {article.get('content_original', '')}".lower()
        match_count = sum(1 for w in search_words if w in text)
        score += 0.4 * (match_count / len(search_words)) if search_words else 0.0

    # 2. Quellen-Reputation (30%)
    source_url = article.get("source_url", "")
    if source_url:
        domain = _extract_domain(source_url)
        category = _detect_category(domain)
        score += 0.3 * CATEGORY_REPUTATION.get(category, 0.4)
    else:
        score += 0.3 * 0.4  # Unbekannte Quelle

    # 3. Inhaltstiefe (20%)
    content = article.get("content_original") or article.get("content_de") or ""
    if len(content) > 500:
        score += 0.2
    elif len(content) > 200:
        score += 0.1
    elif len(content) > 50:
        score += 0.05

    # 4. RSS-Score Bonus (10%)
    score += 0.1 * rss_score

    return min(1.0, score)



async def _verify_article_urls(
    articles: list[dict],
    concurrency: int = 10,
    timeout: float = 8.0,
) -> list[dict]:
    """Prueft WebSearch-URLs auf Erreichbarkeit. Ersetzt unerreichbare URLs durch Suchlinks."""
    if not articles:
        return []

    sem = asyncio.Semaphore(concurrency)
    results: list[dict | None] = [None] * len(articles)

    async def _check(idx: int, article: dict, client: httpx.AsyncClient):
        url = article.get("source_url", "").strip()
        if not url:
            results[idx] = article  # Kein URL -> behalten (wird eh nicht verlinkt)
            return
        async with sem:
            try:
                resp = await client.head(url)
                if resp.status_code == 405:
                    # Manche Server unterstuetzen kein HEAD
                    resp = await client.get(url, headers={"Range": "bytes=0-0"})
                if 200 <= resp.status_code < 400:
                    results[idx] = article
                    return
                # 404 oder anderer Fehler -> Fallback-Suchlink
                logger.info(f"URL-Verifizierung: {resp.status_code} fuer {url}")
            except Exception as e:
                logger.debug(f"URL-Verifizierung fehlgeschlagen fuer {url}: {e}")

            # Fallback: Google-Suchlink aus Headline + Source-Domain
            headline = article.get("headline", "")
            source = article.get("source", "")
            domain = ""
            try:
                from urllib.parse import urlparse as _urlparse
                domain = _urlparse(url).netloc
            except Exception:
                pass
            if headline:
                search_query = f"site:{domain} {headline}" if domain else f"{source} {headline}"
                fallback_url = f"https://www.google.com/search?q={quote_plus(search_query)}"
                article_copy = dict(article)
                article_copy["source_url"] = fallback_url
                article_copy["_url_repaired"] = True
                results[idx] = article_copy
                logger.info(f"URL-Fallback: {url} -> Google-Suche fuer \"{headline[:60]}...\"")
            else:
                results[idx] = article  # Kein Headline -> Original behalten

    async with httpx.AsyncClient(
        timeout=timeout,
        follow_redirects=True,
        headers={"User-Agent": "Mozilla/5.0 (compatible; AegisSight-Monitor/1.0)"},
    ) as client:
        await asyncio.gather(*[_check(i, a, client) for i, a in enumerate(articles)])

    verified = [r for r in results if r is not None]
    repaired = sum(1 for r in verified if r.get("_url_repaired"))
    ok = len(verified) - repaired

    if repaired > 0:
        logger.warning(
            f"URL-Verifizierung: {ok} OK, {repaired} durch Suchlinks ersetzt "
            f"(von {len(articles)} WebSearch-Artikeln)"
        )
    else:
        logger.info(f"URL-Verifizierung: Alle {len(articles)} WebSearch-URLs erreichbar")

    return verified


async def _background_discover_sources(articles: list[dict]):
    """Background-Task: Registriert seriöse, unbekannte Quellen aus Recherche-Ergebnissen."""
    from database import get_db

    db = await get_db()
    try:
        # 1. Unique Domains extrahieren
        seen_domains: set[str] = set()
        domains_to_check: list[tuple[str, str, str]] = []
        for article in articles:
            url = article.get("source_url")
            if not url:
                continue
            domain = _extract_domain(url)
            if not domain or domain in seen_domains:
                continue
            seen_domains.add(domain)

            # 2. Nur seriöse Domains (in DOMAIN_CATEGORY_MAP, nicht "sonstige")
            category = _detect_category(domain)
            if category == "sonstige":
                continue
            domains_to_check.append((domain, url, category))

        if not domains_to_check:
            return

        # 3. Gegen DB prüfen — welche Domains existieren schon?
        new_count = 0
        for domain, url, category in domains_to_check:
            cursor = await db.execute(
                "SELECT id FROM sources WHERE LOWER(domain) = ?",
                (domain.lower(),),
            )
            if await cursor.fetchone():
                continue  # Domain schon bekannt

            # 4. RSS-Feed-Erkennung
            try:
                result = await discover_source(url)
                name = domain_to_display_name(domain)
                source_type = result["source_type"]  # "rss_feed" oder "web_source"
                feed_url = result.get("rss_url")

                await db.execute(
                    """INSERT INTO sources (name, url, domain, source_type, category, status, notes, added_by)
                       VALUES (?, ?, ?, ?, ?, 'active', 'Auto-entdeckt via Recherche', 'system')""",
                    (name, feed_url or f"https://{domain}", domain, source_type, category),
                )
                new_count += 1
                logger.info(f"Neue Quelle registriert: {name} ({domain}) als {source_type}")
            except Exception as e:
                logger.debug(f"Discovery fehlgeschlagen für {domain}: {e}")

        if new_count > 0:
            await db.commit()
            logger.info(f"Background-Discovery: {new_count} neue Quellen registriert")
    except Exception as e:
        logger.warning(f"Background-Discovery Fehler: {e}")
    finally:
        await db.close()


async def _create_notifications_for_incident(
    db, incident_id: int, visibility: str, created_by: int, tenant_id: int, notifications: list[dict]
):
    """Erzeugt DB-Notifications fuer alle betroffenen Nutzer der Organisation.

    - Oeffentliche Lagen -> alle Nutzer der Org
    - Private Lagen -> nur Ersteller
    """
    if not notifications:
        return

    now = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')

    if visibility == "public" and tenant_id:
        cursor = await db.execute(
            "SELECT id FROM users WHERE organization_id = ? AND is_active = 1 AND last_login_at IS NOT NULL",
            (tenant_id,),
        )
        user_ids = [row["id"] for row in await cursor.fetchall()]
    else:
        user_ids = [created_by]

    for user_id in user_ids:
        for notif in notifications:
            await db.execute(
                """INSERT INTO notifications (user_id, incident_id, type, title, text, icon, tenant_id, created_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    user_id,
                    incident_id,
                    notif.get("type", "refresh_summary"),
                    notif["title"],
                    notif["text"],
                    notif.get("icon", "info"),
                    tenant_id,
                    now,
                ),
            )

    await db.commit()
    logger.info(f"Notifications erstellt: {len(notifications)} x {len(user_ids)} Nutzer fuer Lage {incident_id}")


async def _send_email_notifications_for_incident(
    db, incident_id: int, incident_title: str, visibility: str,
    created_by: int, tenant_id: int, notifications: list[dict],
    incident_type: str = "adhoc",
):
    """Sendet E-Mail-Benachrichtigungen basierend auf individuellen Nutzer-Abos.

    Jeder Nutzer hat eigene E-Mail-Praeferenzen pro Lage (incident_subscriptions).
    Nur Nutzer die aktiv sind und sich mindestens einmal eingeloggt haben
    (last_login_at IS NOT NULL) werden beruecksichtigt.
    """
    if not notifications:
        return

    from email_utils.sender import send_email
    from email_utils.templates import incident_notification_email
    from config import MAGIC_LINK_BASE_URL
    from services.org_settings import get_org_language

    # Sprache der Org bestimmen (die Lage gehoert genau einer Org)
    org_lang_iso = await get_org_language(db, tenant_id) if tenant_id else "de"

    # Alle Nutzer mit aktiven Abos fuer diese Lage laden
    cursor = await db.execute(
        """SELECT s.notify_email_summary, s.notify_email_new_articles,
                  s.notify_email_status_change, u.id, u.email, u.username
           FROM incident_subscriptions s
           JOIN users u ON u.id = s.user_id
           WHERE s.incident_id = ?
             AND u.is_active = 1
             AND u.last_login_at IS NOT NULL
             AND (s.notify_email_summary = 1
                  OR s.notify_email_new_articles = 1
                  OR s.notify_email_status_change = 1)""",
        (incident_id,),
    )
    subscribers = await cursor.fetchall()
    if not subscribers:
        return

    dashboard_url = f"{MAGIC_LINK_BASE_URL}/dashboard"

    for sub in subscribers:
        prefs = dict(sub)

        # Relevante Notifications basierend auf Nutzer-Praeferenzen filtern
        filtered_notifications = []
        for n in notifications:
            ntype = n.get("type", "refresh_summary")

            if ntype == "refresh_summary" and prefs.get("notify_email_summary"):
                filtered_notifications.append(n)
            elif ntype == "new_articles" and prefs.get("notify_email_new_articles"):
                filtered_notifications.append(n)
            elif ntype == "status_change" and prefs.get("notify_email_status_change"):
                filtered_notifications.append(n)

        if not filtered_notifications:
            continue

        subject, html = incident_notification_email(
            username=prefs["email"].split("@")[0],
            incident_title=incident_title,
            notifications=filtered_notifications,
            dashboard_url=dashboard_url,
            incident_type=incident_type,
            lang=org_lang_iso,
        )
        try:
            await send_email(prefs["email"], subject, html)
            logger.info(f"E-Mail-Benachrichtigung gesendet an {prefs['email']} fuer Lage {incident_id} ({len(filtered_notifications)} Items)")
        except Exception as e:
            logger.error(f"E-Mail-Benachrichtigung fehlgeschlagen fuer {prefs['email']}: {e}")


class AgentOrchestrator:
    """Koordiniert die Recherche-Zyklen: pro Organisation eine unabhaengige,
    parallel laufende Worker-Lane."""

    # Lane-Schluessel fuer oeffentliche/systemweite Lagen (tenant_id IS NULL).
    PUBLIC_LANE = 0
    # Sekunden ohne Auftrag, nach denen eine Lane sich beendet und aufraeumt.
    IDLE_TIMEOUT = 60

    def __init__(self):
        self._running = False
        self._ws_manager = None
        # Pro Organisation (tenant_id) eine eigene Queue + ein eigener Worker-Task.
        # So arbeiten Organisationen unabhaengig und parallel, statt sich eine
        # globale Warteschlange zu teilen. Lanes werden bei Bedarf angelegt und bei
        # Leerlauf wieder beendet. Der Lock schuetzt das Anlegen/Beenden gegen
        # gleichzeitiges Einreihen.
        self._lanes: dict[int, asyncio.Queue] = {}
        self._lane_workers: dict[int, asyncio.Task] = {}
        self._lanes_lock = asyncio.Lock()
        # Globale Zustaende — incident-IDs sind lane-uebergreifend eindeutig.
        self._queued_ids: set[int] = set()          # eingereiht (in irgendeiner Lane)
        # incident_id -> Session-Start (UTC ISO mit 'Z'). Ersetzt den frueheren
        # Einzelwert; haelt den Frontend-Timer ueber Multi-Pass/Retry stabil.
        self._current_tasks: dict[int, str] = {}
        self._cancel_requested: set[int] = set()
        # incident_id -> Cancel-Event des laufenden Tasks. cancel_refresh laeuft in
        # einem anderen async-Kontext als der Worker und kann die ContextVar nicht
        # nutzen, daher diese direkte Zuordnung.
        self._cancel_events: dict[int, asyncio.Event] = {}
        # Optionales Sicherheitsventil gegen Ueberlastung des gemeinsamen Claude-
        # Kontos: begrenzt die Zahl GLEICHZEITIG laufender Recherchen ueber alle
        # Lanes. Default 0 = unbegrenzt (jede Organisation voellig unabhaengig).
        _max = int(os.getenv("ORCHESTRATOR_MAX_PARALLEL", "0") or "0")
        self._global_sem: Optional[asyncio.Semaphore] = asyncio.Semaphore(_max) if _max > 0 else None

    def set_ws_manager(self, ws_manager):
        """WebSocket-Manager setzen für Echtzeit-Updates."""
        self._ws_manager = ws_manager

    async def start(self):
        """Orchestrator aktivieren. Worker-Lanes werden pro Organisation bei der
        ersten Anfrage angelegt (lazy)."""
        self._running = True
        logger.info("Agenten-Orchestrator gestartet")

    async def stop(self):
        """Orchestrator stoppen und alle Lane-Worker beenden."""
        self._running = False
        async with self._lanes_lock:
            workers = list(self._lane_workers.values())
            self._lane_workers.clear()
            self._lanes.clear()
        for task in workers:
            task.cancel()
        logger.info("Agenten-Orchestrator gestoppt")

    async def enqueue_refresh(self, incident_id: int, trigger_type: str = "manual", user_id: int = None, collect_only: bool = False) -> bool:
        """Refresh-Auftrag in die Lane der Organisation stellen. Gibt False zurueck
        wenn die Lage dort bereits wartet oder gerade laeuft.

        collect_only=True (Studio-Baustein "Sammeln"): sammelt nur Artikel, ueberspringt
        Analyse und Faktencheck.
        """
        if incident_id in self._queued_ids or incident_id in self._current_tasks:
            logger.info(f"Refresh fuer Lage {incident_id} uebersprungen: bereits aktiv/in Queue")
            return False

        visibility, created_by, tenant_id = await self._get_incident_visibility(incident_id)
        # Doppelspur (EU-Umbau Phase 2). Eine Lane je Organisation UND KI-Weg.
        # Ein CLI-Lauf und ein EU-Lauf derselben Organisation laufen dadurch
        # wirklich gleichzeitig (getrennte Ressourcen, Abo gegen AWS), während
        # Läufe über denselben Weg seriell bleiben (Rate-Limit-Schutz).
        ai_backend = await self._resolve_ai_backend(incident_id)
        lane_key = (tenant_id if tenant_id else self.PUBLIC_LANE, ai_backend)

        async with self._lanes_lock:
            queue = self._lanes.get(lane_key)
            if queue is None:
                queue = asyncio.Queue()
                self._lanes[lane_key] = queue
                self._lane_workers[lane_key] = asyncio.create_task(self._worker(lane_key, queue))
                logger.info(f"Neue Worker-Lane fuer Organisation {lane_key} gestartet")
            self._queued_ids.add(incident_id)
            queue.put_nowait((incident_id, trigger_type, user_id, collect_only))
            queue_size = queue.qsize()
        logger.info(f"Refresh fuer Lage {incident_id} eingereiht (Lane {lane_key}, Queue: {queue_size}, Trigger: {trigger_type})")

        if self._ws_manager:
            await self._ws_manager.broadcast_for_incident({
                "type": "status_update",
                "incident_id": incident_id,
                "data": {"status": "queued", "queue_position": queue_size},
            }, visibility, created_by, tenant_id)

        return True

    async def cancel_refresh(self, incident_id: int) -> bool:
        """Fordert Abbruch eines laufenden oder wartenden Refreshes an."""
        # Laeuft die Lage gerade?
        if incident_id in self._current_tasks:
            self._cancel_requested.add(incident_id)
            ev = self._cancel_events.get(incident_id)
            if ev:
                ev.set()
            logger.info(f"Cancel angefordert fuer laufende Lage {incident_id}")
            if self._ws_manager:
                try:
                    vis, cb, tid = await self._get_incident_visibility(incident_id)
                except Exception:
                    vis, cb, tid = "public", None, None
                await self._ws_manager.broadcast_for_incident({
                    "type": "status_update",
                    "incident_id": incident_id,
                    "data": {"status": "cancelling", "detail": "Wird abgebrochen..."},
                }, vis, cb, tid)
            return True

        # Wartet die Lage noch in ihrer Lane?
        if incident_id in self._queued_ids:
            self._queued_ids.discard(incident_id)
            # Betroffene Lane bestimmen und die ID aus deren Queue entfernen.
            try:
                _vis, _cb, _tid = await self._get_incident_visibility(incident_id)
            except Exception:
                _tid = None
            lane_key = _tid if _tid else self.PUBLIC_LANE
            removed = False
            async with self._lanes_lock:
                queue = self._lanes.get(lane_key)
                if queue is not None:
                    new_items = []
                    while not queue.empty():
                        try:
                            item = queue.get_nowait()
                        except Exception:
                            break
                        iid = item[0] if isinstance(item, tuple) else item
                        if iid == incident_id:
                            removed = True
                            queue.task_done()
                        else:
                            new_items.append(item)
                    for item in new_items:
                        queue.put_nowait(item)

            logger.info(f"Lage {incident_id} aus Warteschlange entfernt (removed={removed})")

            # refresh_log-Eintrag schreiben, damit Auto-Refresh nicht im naechsten Tick erneut einreiht
            await self._log_queued_cancellation(incident_id)

            # Send cancelled event
            if self._ws_manager:
                try:
                    vis, cb, tid = await self._get_incident_visibility(incident_id)
                except Exception:
                    vis, cb, tid = "public", None, None
                await self._ws_manager.broadcast_for_incident({
                    "type": "refresh_cancelled",
                    "incident_id": incident_id,
                    "data": {"status": "cancelled"},
                }, vis, cb, tid)
            return True

        return False

    def _check_cancelled(self, incident_id: int):
        """Prüft ob Abbruch angefordert wurde und wirft CancelledError."""
        if incident_id in self._cancel_requested:
            self._cancel_requested.discard(incident_id)
            raise asyncio.CancelledError("Vom Nutzer abgebrochen")

    async def _worker(self, lane_key: int, queue: asyncio.Queue):
        """Verarbeitet die Auftraege EINER Organisation sequentiell. Verschiedene
        Organisationen laufen in eigenen Lanes parallel. Bei Leerlauf beendet sich
        die Lane selbst und wird beim naechsten Auftrag neu angelegt."""
        while self._running:
            try:
                item = await asyncio.wait_for(queue.get(), timeout=self.IDLE_TIMEOUT)
            except asyncio.TimeoutError:
                # Leerlauf: Lane beenden. Unter Lock gegen gleichzeitiges Einreihen,
                # damit kein Auftrag in einer verwaisten Queue liegen bleibt.
                async with self._lanes_lock:
                    if queue.empty() and self._lanes.get(lane_key) is queue:
                        self._lanes.pop(lane_key, None)
                        self._lane_workers.pop(lane_key, None)
                        logger.info(f"Worker-Lane fuer Organisation {lane_key} bei Leerlauf beendet")
                        return
                continue

            if len(item) == 4:
                incident_id, trigger_type, user_id, collect_only = item
            elif len(item) == 3:
                incident_id, trigger_type, user_id = item
                collect_only = False
            else:
                incident_id, trigger_type = item
                user_id = None
                collect_only = False
            self._queued_ids.discard(incident_id)
            # Session-Start EINMAL setzen — bleibt ueber Multi-Pass/Retry hinweg stabil
            self._current_tasks[incident_id] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
            cancel_event = asyncio.Event()
            self._cancel_events[incident_id] = cancel_event
            _cancel_event_var.set(cancel_event)
            logger.info(f"Starte Refresh für Lage {incident_id} (Lane {lane_key}, Trigger: {trigger_type})")

            RETRY_DELAYS = [0, 120, 300]  # Sekunden: sofort, 2min, 5min
            TRANSIENT_ERRORS = (asyncio.TimeoutError, TimeoutError, ConnectionError, OSError)
            from agents.claude_client import ClaudeCliError
            last_error = None

            def _is_transient_cli(err: Exception) -> bool:
                return isinstance(err, ClaudeCliError) and err.error_type in ("rate_limit", "timeout")

            def _is_db_locked(err: Exception) -> bool:
                # Seit der Doppelspur (Phase 2) koennen sich parallele Lanes beim
                # Schreiben auf die SQLite-DB kurz blockieren. Das ist transient
                # und einen Retry wert, kein permanenter Fehler.
                import sqlite3 as _sqlite3
                return isinstance(err, _sqlite3.OperationalError) and "locked" in str(err).lower()

            # Optionales globales Ventil (Default aus): begrenzt gleichzeitige Recherchen.
            sem = self._global_sem
            if sem is not None:
                await sem.acquire()
            try:
                # Research-Lagen: Automatisch 3 Durchläufe nur beim ersten Refresh
                incident_type, has_summary = await self._get_incident_info(incident_id)
                use_multi_pass = incident_type == "research" and not has_summary and not collect_only

                for attempt in range(3):
                    try:
                        if use_multi_pass:
                            await self._run_research_multi_pass(incident_id, trigger_type=trigger_type, user_id=user_id)
                        else:
                            await self._run_refresh(incident_id, trigger_type=trigger_type, retry_count=attempt, user_id=user_id, collect_only=collect_only)
                        last_error = None
                        break  # Erfolg
                    except asyncio.CancelledError:
                        logger.info(f"Refresh fuer Lage {incident_id} abgebrochen")
                        await self._mark_refresh_cancelled(incident_id)
                        try:
                            _vis, _cb, _tid = await self._get_incident_visibility(incident_id)
                        except Exception:
                            _vis, _cb, _tid = "public", None, None
                        if self._ws_manager:
                            await self._ws_manager.broadcast_for_incident({
                                "type": "refresh_cancelled",
                                "incident_id": incident_id,
                                "data": {"status": "cancelled"},
                            }, _vis, _cb, _tid)
                        last_error = None
                        break
                    except Exception as e:
                        # Auth/CLI-Fehler: sofort abbrechen, kein Retry sinnvoll
                        if isinstance(e, ClaudeCliError) and e.error_type in ("auth_error", "cli_error"):
                            last_error = e
                            logger.error(f"Permanenter Claude-Fehler [{e.error_type}] bei Lage {incident_id}: {e}")
                            await self._mark_refresh_failed(incident_id, str(e))
                            break

                        # Transiente Fehler: Retry bis 3x
                        if isinstance(e, TRANSIENT_ERRORS) or _is_transient_cli(e) or _is_db_locked(e):
                            last_error = e
                            kind = e.error_type if isinstance(e, ClaudeCliError) else type(e).__name__
                            logger.warning(f"Transienter Fehler [{kind}] bei Lage {incident_id} (Versuch {attempt + 1}/3): {e}")
                            if attempt < 2:
                                await self._mark_refresh_failed(incident_id, str(e))
                                delay = RETRY_DELAYS[attempt + 1]
                                logger.info(f"Retry in {delay}s für Lage {incident_id}")
                                if self._ws_manager:
                                    try:
                                        _vis, _cb, _tid = await self._get_incident_visibility(incident_id)
                                    except Exception:
                                        _vis, _cb, _tid = "public", None, None
                                    await self._ws_manager.broadcast_for_incident({
                                        "type": "status_update",
                                        "incident_id": incident_id,
                                        "data": {"status": "retrying", "attempt": attempt + 1, "delay": delay},
                                    }, _vis, _cb, _tid)
                                await asyncio.sleep(delay)
                                continue
                            else:
                                await self._mark_refresh_failed(incident_id, f"Endgültig fehlgeschlagen nach 3 Versuchen: {e}")
                                break

                        # Alles andere: permanent
                        last_error = e
                        logger.error(f"Permanenter Fehler bei Refresh für Lage {incident_id}: {e}")
                        await self._mark_refresh_failed(incident_id, str(e))
                        break

                if last_error and self._ws_manager:
                    try:
                        _vis, _cb, _tid = await self._get_incident_visibility(incident_id)
                    except Exception:
                        _vis, _cb, _tid = "public", None, None
                    await self._ws_manager.broadcast_for_incident({
                        "type": "refresh_error",
                        "incident_id": incident_id,
                        "data": {"error": str(last_error)},
                    }, _vis, _cb, _tid)
            finally:
                if sem is not None:
                    sem.release()
                self._current_tasks.pop(incident_id, None)
                self._cancel_events.pop(incident_id, None)
                self._cancel_requested.discard(incident_id)
                _cancel_event_var.set(None)
                _ai_backend_var.set(None)
                queue.task_done()

    async def _mark_refresh_cancelled(self, incident_id: int):
        """Markiert den laufenden Refresh-Log-Eintrag als cancelled und schliesst
        alle noch aktiven Pipeline-Schritte. Ohne den zweiten Schritt blieb der
        zuletzt aktive Step-Eintrag verwaist und das Frontend zeigte dauerhaft
        'Schritt X laeuft', weil /api/incidents/<id>/pipeline aus
        refresh_pipeline_steps liest."""
        from database import get_db
        from services.pipeline_tracker import cancel_active_steps
        db = await get_db()
        try:
            now_str = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')
            cur = await db.execute(
                "SELECT id FROM refresh_log WHERE incident_id = ? AND status = 'running'",
                (incident_id,),
            )
            row = await cur.fetchone()
            refresh_log_id = row["id"] if row else None

            await db.execute(
                """UPDATE refresh_log SET status = 'cancelled', error_message = 'Vom Nutzer abgebrochen',
                   completed_at = ? WHERE incident_id = ? AND status = 'running'""",
                (now_str, incident_id),
            )
            await db.commit()

            if refresh_log_id is not None:
                await cancel_active_steps(db, refresh_log_id=refresh_log_id)
        except Exception as e:
            logger.warning(f"Konnte Refresh-Log nicht als abgebrochen markieren: {e}")
        finally:
            await db.close()

    async def _log_queued_cancellation(self, incident_id: int):
        """Schreibt einen cancelled-Eintrag fuer einen Queue-Abbruch (Lage war noch nicht laufend).
        Verhindert, dass der Auto-Refresh-Scheduler im naechsten Tick sofort wieder einreiht."""
        from database import get_db
        db = await get_db()
        try:
            cur = await db.execute("SELECT tenant_id FROM incidents WHERE id = ?", (incident_id,))
            row = await cur.fetchone()
            tid = row["tenant_id"] if row else None
            now_str = datetime.now(TIMEZONE).strftime("%Y-%m-%d %H:%M:%S")
            await db.execute(
                """INSERT INTO refresh_log (incident_id, started_at, completed_at, status,
                   trigger_type, error_message, tenant_id)
                   VALUES (?, ?, ?, 'cancelled', 'manual', 'Aus Warteschlange entfernt', ?)""",
                (incident_id, now_str, now_str, tid),
            )
            await db.commit()
        except Exception as e:
            logger.warning(f"Konnte Queue-Cancel nicht in refresh_log loggen: {e}")
        finally:
            await db.close()

    async def _mark_refresh_failed(self, incident_id: int, error: str):
        """Markiert den laufenden Refresh-Log-Eintrag als error."""
        from database import get_db
        db = await get_db()
        try:
            await db.execute(
                """UPDATE refresh_log SET status = 'error', error_message = ?,
                   completed_at = ? WHERE incident_id = ? AND status = 'running'""",
                (error[:500], datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S'), incident_id),
            )
            await db.commit()
        except Exception as e:
            logger.warning(f"Konnte Refresh-Log nicht als fehlgeschlagen markieren: {e}")
        finally:
            await db.close()

    async def _get_incident_visibility(self, incident_id: int) -> tuple[str, Optional[int], Optional[int]]:
        """Incident-Visibility, created_by und tenant_id laden."""
        from database import get_db
        visibility = "public"
        created_by = None
        tenant_id = None
        db = await get_db()
        try:
            cursor = await db.execute(
                "SELECT visibility, created_by, tenant_id FROM incidents WHERE id = ?", (incident_id,)
            )
            row = await cursor.fetchone()
            if row:
                visibility = row["visibility"] or "public"
                created_by = row["created_by"]
                tenant_id = row["tenant_id"]
        finally:
            await db.close()
        return visibility, created_by, tenant_id

    async def _resolve_ai_backend(self, incident_id: int) -> str:
        """Löst das KI-Backend einer Lage auf (EU-Umbau).

        Lage (incidents.ai_backend) gewinnt vor dem Org-Setting 'ai_backend',
        das vor dem globalen Default AI_BACKEND aus config. Wird beim Einreihen
        (Lane-Wahl) und beim Refresh-Start (ContextVar) identisch genutzt."""
        from config import AI_BACKEND
        from database import get_db
        from services.org_settings import get_org_setting
        incident_backend = None
        tenant_id = None
        db = await get_db()
        try:
            cursor = await db.execute(
                "SELECT ai_backend, tenant_id FROM incidents WHERE id = ?", (incident_id,)
            )
            row = await cursor.fetchone()
            if row:
                incident_backend = row["ai_backend"] if "ai_backend" in row.keys() else None
                tenant_id = row["tenant_id"] if "tenant_id" in row.keys() else None
            org_backend = await get_org_setting(db, tenant_id, "ai_backend", default=None) if tenant_id else None
        finally:
            await db.close()
        ai_backend = (incident_backend or org_backend or AI_BACKEND or "cli").strip().lower()
        if ai_backend not in ("cli", "bedrock"):
            logger.warning("Unbekanntes ai_backend '%s' für Lage %s, fallback 'cli'", ai_backend, incident_id)
            ai_backend = "cli"
        return ai_backend

    async def _run_refresh(self, incident_id: int, trigger_type: str = "manual", retry_count: int = 0, user_id: int = None, _suppress_complete: bool = False, _pass_info: dict = None, collect_only: bool = False):
        """Führt einen kompletten Refresh-Zyklus durch."""
        import aiosqlite
        from database import get_db
        from agents.researcher import ResearcherAgent
        from agents.analyzer import AnalyzerAgent
        from agents.factchecker import FactCheckerAgent
        from feeds.rss_parser import RSSParser
        from services import pipeline_tracker as _pipe

        db = await get_db()
        try:
            # Lage laden
            cursor = await db.execute("SELECT * FROM incidents WHERE id = ?", (incident_id,))
            incident = await cursor.fetchone()
            if not incident:
                logger.warning(f"Lage {incident_id} nicht gefunden")
                return

            title = incident["title"]
            description = incident["description"] or ""
            incident_type = incident["type"] or "adhoc"
            international = bool(incident["international_sources"]) if "international_sources" in incident.keys() else True
            # Wenn die Org eine Sprach-Whitelist gesetzt hat, ist 'international' bedeutungslos —
            # die Whitelist gewinnt. Wir setzen 'international' auf True, damit der nachgelagerte
            # Code alle (durch Whitelist gefilterten) Feeds in Betracht zieht. Tatsaechliche
            # Einschraenkung passiert in get_feeds_with_metadata.
            # Hinweis: source_lang_whitelist wird weiter unten geladen.
            include_telegram = bool(incident["include_telegram"]) if "include_telegram" in incident.keys() else False
            include_x = bool(incident["include_x"]) if "include_x" in incident.keys() else False
            visibility = incident["visibility"] if "visibility" in incident.keys() else "public"
            created_by = incident["created_by"] if "created_by" in incident.keys() else None
            tenant_id = incident["tenant_id"] if "tenant_id" in incident.keys() else None
            # Org-Sprache fuer alle KI-Agenten (Lagebild, Faktencheck, Recherche)
            from services.org_settings import (
                get_org_language, language_display, get_research_language,
                get_source_language_whitelist, get_translator_enabled,
            )
            # KI-Backend für diesen Refresh auflösen und per ContextVar setzen
            # (EU-Umbau, gemeinsame Auflösung mit der Lane-Wahl beim Einreihen,
            # siehe _resolve_ai_backend).
            ai_backend = await self._resolve_ai_backend(incident_id)
            _ai_backend_var.set(ai_backend)
            if ai_backend != "cli":
                logger.info("Lage %s läuft über KI-Backend '%s'", incident_id, ai_backend)
            # Wartebudget des EU-Wegs zuruecksetzen und die Oberflaeche an die
            # Wartemeldungen anschliessen. Ohne diesen Hinweis sieht eine
            # Wiederholung nach einem Kapazitaetsengpass wie ein Haenger aus,
            # weil die Anzeige waehrend der Wartezeit stillsteht.
            bedrock_client.refresh_kontext_starten()
            if self._ws_manager:
                def _warte_melden(art: str, sekunden: float, _iid=incident_id,
                                  _vis=visibility, _cb=created_by, _tid=tenant_id) -> None:
                    text = bedrock_client._ART_TEXT.get(art, art)
                    asyncio.create_task(self._ws_manager.broadcast_for_incident({
                        "type": "status_update",
                        "incident_id": _iid,
                        "data": {
                            "status": "running",
                            "detail": f"KI-Dienst {text}, neuer Versuch in {sekunden:.0f} Sekunden...",
                        },
                    }, _vis, _cb, _tid))
                bedrock_client.wartemelder_setzen(_warte_melden)
            output_language_iso = await get_org_language(db, tenant_id) if tenant_id else "de"
            output_language = language_display(output_language_iso)
            # research_language steuert nur den WebSearch-Prompt ("suche in Sprache X").
            # Default = output_language_iso. Bei jp_demo wird das auf 'ja' gesetzt, waehrend
            # output_language_iso 'de' bleibt (Lagebild auf Deutsch, Recherche auf Japanisch).
            research_language_iso = await get_research_language(db, tenant_id) if tenant_id else output_language_iso
            # source_language_whitelist schraenkt RSS-/Telegram-Quellenpool ein (z.B. ['ja']).
            # Wenn gesetzt, wird das incident-level Flag international_sources ignoriert
            # (Whitelist ist explizit, das Flag ist Default-Verhalten).
            source_lang_whitelist = await get_source_language_whitelist(db, tenant_id) if tenant_id else None
            # Pro-Org-Override des globalen TRANSLATOR_ENABLED-Flags.
            translator_enabled = await get_translator_enabled(db, tenant_id)
            # Whitelist gewinnt ueber das incident-Flag international_sources:
            # wenn die Org eine Sprach-Whitelist hat, sind alle gewaehlten Feeds
            # ohnehin "Wunsch-Sprache" — kein Splitting in primary/international noetig.
            if source_lang_whitelist:
                international = True
                logger.info(
                    "Org %s hat source_language_whitelist=%s gesetzt; "
                    "incident.international_sources wird ignoriert",
                    tenant_id, source_lang_whitelist,
                )
            previous_summary = incident["summary"] or ""
            previous_sources_json = incident["sources_json"] if "sources_json" in incident.keys() else None
            previous_developments = incident["latest_developments"] if "latest_developments" in incident.keys() else None

            # Bei Retry: vorherigen running-Eintrag als error markieren
            if retry_count > 0:
                await db.execute(
                    """UPDATE refresh_log SET status = 'error', error_message = 'Retry gestartet',
                       completed_at = ? WHERE incident_id = ? AND status = 'running'""",
                    (datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S'), incident_id),
                )
                await db.commit()

            # Refresh-Log starten
            now = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')
            now_utc = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
            cursor = await db.execute(
                "INSERT INTO refresh_log (incident_id, started_at, status, trigger_type, retry_count, tenant_id) VALUES (?, ?, 'running', ?, ?, ?)",
                (incident_id, now, trigger_type, retry_count, tenant_id),
            )
            await db.commit()
            log_id = cursor.lastrowid
            usage_acc = UsageAccumulator()

            # --- Pipeline-Tracking (Analysepipeline-Visualisierung) ---
            _pass_nr = (_pass_info or {}).get("nr", 1)
            _step_ids: dict[str, Optional[int]] = {}

            async def _pipe_start(step_key: str):
                try:
                    sid = await _pipe.start_step(
                        db, self._ws_manager,
                        refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
                        pass_number=_pass_nr, tenant_id=tenant_id,
                        visibility=visibility, created_by=created_by,
                    )
                    _step_ids[step_key] = sid
                    return sid
                except Exception as _e:
                    logger.debug(f"_pipe_start({step_key}) ignoriert: {_e}")
                    return None

            async def _pipe_done(step_key: str, count_value=None, count_secondary=None):
                try:
                    sid = _step_ids.pop(step_key, None)
                    await _pipe.complete_step(
                        db, self._ws_manager, step_id=sid,
                        refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
                        pass_number=_pass_nr, count_value=count_value, count_secondary=count_secondary,
                        tenant_id=tenant_id, visibility=visibility, created_by=created_by,
                    )
                except Exception as _e:
                    logger.debug(f"_pipe_done({step_key}) ignoriert: {_e}")

            async def _pipe_skip(step_key: str):
                try:
                    await _pipe.skip_step(
                        db, self._ws_manager,
                        refresh_log_id=log_id, incident_id=incident_id, step_key=step_key,
                        pass_number=_pass_nr, tenant_id=tenant_id,
                        visibility=visibility, created_by=created_by,
                    )
                except Exception as _e:
                    logger.debug(f"_pipe_skip({step_key}) ignoriert: {_e}")

            research_status = "deep_researching" if incident_type == "research" else "researching"
            research_detail = "Hintergrundrecherche im Web läuft..." if incident_type == "research" else "RSS-Feeds und Web werden durchsucht..."
            # Multi-Pass: Detail-Text mit Durchlauf-Info versehen
            _ws_extra = {}
            if _pass_info:
                _pnr, _ptotal, _plabel = _pass_info["nr"], _pass_info["total"], _pass_info["label"]
                research_detail = f"Recherche {_pnr}/{_ptotal}: {_plabel}..."
                _ws_extra = {"research_pass": _pnr, "research_total_passes": _ptotal}
            if self._ws_manager:
                await self._ws_manager.broadcast_for_incident({
                    "type": "status_update",
                    "incident_id": incident_id,
                    "data": {"status": research_status, "detail": research_detail, "started_at": now_utc, **_ws_extra},
                }, visibility, created_by, tenant_id)

            # Bestehende Artikel vorladen (für Dedup UND Kontext)
            cursor = await db.execute(
                "SELECT id, source_url, headline, source FROM articles WHERE incident_id = ?",
                (incident_id,),
            )
            existing_db_articles_full = await cursor.fetchall()

            # Pipeline-Schritt 1: Quellen sichten (vorbereitet)
            await _pipe_start("sources_review")
            try:
                if incident_type == "adhoc":
                    _src_cursor = await db.execute(
                        "SELECT COUNT(*) AS cnt FROM sources "
                        "WHERE status = 'active' "
                        "AND (tenant_id IS NULL OR tenant_id = ?)",
                        (tenant_id,),
                    )
                    _src_row = await _src_cursor.fetchone()
                    _src_total = _src_row["cnt"] if _src_row else 0
                else:
                    _src_total = None
            except Exception:
                _src_total = None
            # secondary wird später mit der Anzahl tatsächlich liefernder Quellen ergänzt
            await _pipe_done("sources_review", count_value=_src_total, count_secondary=None)

            # Schritt 1+2: RSS-Feeds und Claude-Recherche parallel ausführen
            async def _rss_pipeline():
                """RSS-Feed-Suche (Feed-Selektion + dynamische Keywords + Parsing)."""
                if incident_type != "adhoc":
                    logger.info("Recherche-Modus: RSS-Feeds übersprungen")
                    return [], None

                rss_researcher = ResearcherAgent()
                rss_parser = RSSParser()

                from source_rules import get_feeds_with_metadata
                all_feeds = await get_feeds_with_metadata(tenant_id=tenant_id)

                # Dynamische Keywords aus den letzten Headlines extrahieren
                cursor_hl = await db.execute(
                    """SELECT COALESCE(headline_de, headline) as hl
                       FROM articles WHERE incident_id = ?
                       AND COALESCE(headline_de, headline) IS NOT NULL
                       ORDER BY collected_at DESC LIMIT 30""",
                    (incident_id,),
                )
                recent_headlines = [row["hl"] for row in await cursor_hl.fetchall() if row["hl"]]
                dynamic_keywords, kw_usage = await rss_researcher.extract_dynamic_keywords(title, recent_headlines)
                if kw_usage:
                    usage_acc.add(kw_usage)

                feed_usage = None
                keywords = dynamic_keywords  # Dynamische Keywords bevorzugen
                if len(all_feeds) > 20:
                    selected_feeds, feed_sel_keywords, feed_usage = await rss_researcher.select_relevant_feeds(
                        title, description, international, all_feeds
                    )
                    logger.info(f"Feed-Selektion: {len(selected_feeds)} von {len(all_feeds)} Feeds ausgewählt")
                    # Feed-Selektion-Keywords nur als Fallback wenn dynamische fehlen
                    if not keywords:
                        keywords = feed_sel_keywords
                    # --- Recall-Boost: dynamische Google-News-Volltext-Suchfeeds ---
                    # Statt nur feste site:-Feeds zu durchsuchen, baut die Pipeline
                    # pro Sprache einen Google-News-Suchfeed aus den Keywords. Damit
                    # erreichen wir Quellen, die in keinem festen Feed stehen
                    # (Vendor-Blogs, Fachportale, Regionalmedien).
                    from agents.researcher import build_news_search_feeds
                    if source_lang_whitelist:
                        _gnews_langs = list(source_lang_whitelist)
                    else:
                        _gnews_langs = list({output_language_iso, research_language_iso})
                    # Zwei Sets: ein Kontext-Feed (alle Zeiten) + ein Frische-Feed
                    # (when:14d). Der Frische-Feed garantiert, dass das aktuelle
                    # Bild eingefangen wird, auch wenn aeltere Artikel relevanter
                    # ranken. Beide laufen durch dieselbe Pipeline; Dedup entfernt
                    # Ueberschneidungen.
                    _gnews_feeds = build_news_search_feeds(keywords, _gnews_langs)
                    _gnews_recent = build_news_search_feeds(keywords, _gnews_langs, recency_days=14)
                    _all_gnews = _gnews_feeds + _gnews_recent
                    if _all_gnews:
                        logger.info(
                            f"Google-News-Suchfeeds ergaenzt: {len(_gnews_feeds)} Kontext "
                            f"+ {len(_gnews_recent)} Frische (when:14d)"
                        )
                    articles = await rss_parser.search_feeds_selective(
                        title, selected_feeds + _all_gnews, keywords=keywords,
                    )
                else:
                    articles = await rss_parser.search_feeds(title, international=international, tenant_id=tenant_id, keywords=keywords, user_id=user_id)

                logger.info(f"RSS: {len(articles)} relevante Artikel gefunden (international={international})")
                return articles, feed_usage

            async def _web_search_pipeline():
                """Claude WebSearch-Recherche mit Vorselektion eingetragener Web-Quellen."""
                researcher = ResearcherAgent()
                # Bestehende Artikel als Kontext mitgeben (Research + Adhoc)
                existing_for_context = None
                if existing_db_articles_full:
                    existing_for_context = [
                        {"source": row["source"] if "source" in row.keys() else "",
                         "headline": row["headline"],
                         "source_url": row["source_url"]}
                        for row in existing_db_articles_full
                    ]

                # Web-Quellen vorselektieren (Haiku) — nur thematisch passende werden Claude im Prompt empfohlen
                preferred_sources = []
                try:
                    from source_rules import get_feeds_with_metadata
                    web_sources = await get_feeds_with_metadata(tenant_id=tenant_id, source_type="web_source")
                    if web_sources:
                        preferred_sources, web_sel_usage = await researcher.select_relevant_web_sources(
                            title, description, web_sources,
                        )
                        if web_sel_usage:
                            usage_acc.add(web_sel_usage)
                except Exception as e:
                    logger.warning(f"Web-Source-Vorselektion fehlgeschlagen (Pipeline laeuft weiter): {e}")
                    preferred_sources = []

                results, usage, parse_failed = await researcher.search(
                    title, description, incident_type,
                    international=international, user_id=user_id,
                    existing_articles=existing_for_context,
                    preferred_sources=preferred_sources,
                    output_language=output_language,
                    output_language_iso=output_language_iso,
                    research_language_iso=research_language_iso,
                )
                logger.info(
                    f"Claude-Recherche: {len(results)} Ergebnisse"
                    + (f" (mit {len(preferred_sources)} Web-Quellen-Hinweis)" if preferred_sources else "")
                    + (" (Parser fehlgeschlagen)" if parse_failed else "")
                )
                return results, usage, parse_failed

            async def _podcast_pipeline():
                """Podcast-Episoden-Suche (nur adhoc-Lagen, nur mit vorhandenen Transkripten)."""
                if incident_type != "adhoc":
                    logger.info("Recherche-Modus: Podcasts uebersprungen")
                    return [], None

                from source_rules import get_feeds_with_metadata
                podcast_feeds = await get_feeds_with_metadata(tenant_id=tenant_id, source_type="podcast_feed")
                if not podcast_feeds:
                    return [], None

                from feeds.podcast_parser import PodcastFeedParser
                pd_parser = PodcastFeedParser()
                pd_researcher = ResearcherAgent()

                # Dynamische Keywords (eigener Haiku-Call, parallel zu RSS —
                # billig und hält Pipelines unabhaengig)
                cursor_pd_hl = await db.execute(
                    """SELECT COALESCE(headline_de, headline) as hl
                       FROM articles WHERE incident_id = ?
                       AND COALESCE(headline_de, headline) IS NOT NULL
                       ORDER BY collected_at DESC LIMIT 30""",
                    (incident_id,),
                )
                pd_headlines = [row["hl"] for row in await cursor_pd_hl.fetchall() if row["hl"]]
                pd_keywords, pd_kw_usage = await pd_researcher.extract_dynamic_keywords(title, pd_headlines)
                if pd_kw_usage:
                    usage_acc.add(pd_kw_usage)

                # Podcast-Parser erwartet (noch) eine flache Liste – Podcasts sind
                # primaer deutschsprachig, daher reicht das gemeinsame Flatten.
                from agents.researcher import flatten_keywords
                pd_keywords_flat = flatten_keywords(pd_keywords)
                articles = await pd_parser.search_feeds_selective(title, podcast_feeds, keywords=pd_keywords_flat or None)
                logger.info(f"Podcast-Pipeline: {len(articles)} Episoden gefunden")
                return articles, None

            async def _telegram_pipeline():
                """Telegram-Kanal-Suche mit KI-basierter Kanal-Selektion."""
                from feeds.telegram_parser import TelegramParser
                tg_parser = TelegramParser()

                # Alle Telegram-Kanaele laden
                all_channels = await tg_parser._get_telegram_channels(tenant_id=tenant_id)
                if not all_channels:
                    logger.info("Keine Telegram-Kanaele konfiguriert")
                    return [], None

                # KI waehlt relevante Kanaele aus
                tg_researcher = ResearcherAgent()
                selected_channels, tg_sel_usage = await tg_researcher.select_relevant_telegram_channels(
                    title, description, all_channels
                )
                if tg_sel_usage:
                    usage_acc.add(tg_sel_usage)

                selected_ids = [ch["id"] for ch in selected_channels]
                logger.info(f"Telegram-Selektion: {len(selected_ids)} von {len(all_channels)} Kanaelen")

                # Dynamische Keywords fuer Telegram (eigener Aufruf, da parallel zu RSS)
                cursor_tg_hl = await db.execute(
                    """SELECT COALESCE(headline_de, headline) as hl
                       FROM articles WHERE incident_id = ?
                       AND COALESCE(headline_de, headline) IS NOT NULL
                       ORDER BY collected_at DESC LIMIT 30""",
                    (incident_id,),
                )
                tg_headlines = [row["hl"] for row in await cursor_tg_hl.fetchall() if row["hl"]]
                tg_keywords, tg_kw_usage = await tg_researcher.extract_dynamic_keywords(title, tg_headlines)
                if tg_kw_usage:
                    usage_acc.add(tg_kw_usage)
                if isinstance(tg_keywords, dict):
                    logger.info(f"Telegram-Keywords (Sprachen): { {k: len(v) for k, v in tg_keywords.items()} }")
                else:
                    logger.info(f"Telegram-Keywords: {tg_keywords}")

                articles = await tg_parser.search_channels(title, tenant_id=tenant_id, keywords=tg_keywords, channel_ids=selected_ids)
                logger.info(f"Telegram-Pipeline: {len(articles)} Nachrichten")
                return articles, None

            async def _x_pipeline():
                """X-Account-Suche (Twitter) mit KI-basierter Account-Selektion."""
                from feeds.x_parser import XParser
                x_parser = XParser()

                # Alle X-Accounts laden
                all_accounts = await x_parser._get_x_accounts(tenant_id=tenant_id)
                if not all_accounts:
                    logger.info("Keine X-Accounts konfiguriert")
                    return [], None

                # KI waehlt relevante Accounts aus
                x_researcher = ResearcherAgent()
                selected_accounts, x_sel_usage = await x_researcher.select_relevant_x_accounts(
                    title, description, all_accounts
                )
                if x_sel_usage:
                    usage_acc.add(x_sel_usage)

                selected_ids = [acc["id"] for acc in selected_accounts]
                logger.info(f"X-Selektion: {len(selected_ids)} von {len(all_accounts)} Accounts")

                # Dynamische Keywords fuer X (eigener Aufruf, da parallel zu RSS)
                cursor_x_hl = await db.execute(
                    """SELECT COALESCE(headline_de, headline) as hl
                       FROM articles WHERE incident_id = ?
                       AND COALESCE(headline_de, headline) IS NOT NULL
                       ORDER BY collected_at DESC LIMIT 30""",
                    (incident_id,),
                )
                x_headlines = [row["hl"] for row in await cursor_x_hl.fetchall() if row["hl"]]
                x_keywords, x_kw_usage = await x_researcher.extract_dynamic_keywords(title, x_headlines)
                if x_kw_usage:
                    usage_acc.add(x_kw_usage)

                articles = await x_parser.search_accounts(
                    title, tenant_id=tenant_id, keywords=x_keywords, account_ids=selected_ids
                )
                logger.info(f"X-Pipeline: {len(articles)} Posts")
                return articles, None

            # Pipeline-Schritt 2: Nachrichten sammeln (Start)
            await _pipe_start("collect")

            # Pipelines parallel starten (RSS + WebSearch + Podcasts + optional Telegram/X)
            pipelines = [_rss_pipeline(), _web_search_pipeline(), _podcast_pipeline()]
            telegram_idx = x_idx = None
            if include_telegram:
                telegram_idx = len(pipelines)
                pipelines.append(_telegram_pipeline())
            if include_x:
                x_idx = len(pipelines)
                pipelines.append(_x_pipeline())

            pipeline_results = await asyncio.gather(*pipelines)

            (rss_articles, rss_feed_usage) = pipeline_results[0]
            (search_results, search_usage, search_parse_failed) = pipeline_results[1]
            (podcast_articles, _podcast_usage) = pipeline_results[2]
            telegram_articles = pipeline_results[telegram_idx][0] if telegram_idx is not None else []
            x_articles = pipeline_results[x_idx][0] if x_idx is not None else []

            # Podcast-Artikel in die RSS-Liste einfuegen (gleicher Downstream-Pfad)
            if podcast_articles:
                rss_articles = (rss_articles or []) + podcast_articles

            # URL-Verifizierung nur fuer WebSearch-Ergebnisse (RSS-URLs sind bereits verifiziert)
            if search_results:
                search_results = await _verify_article_urls(search_results)

            if rss_feed_usage:
                usage_acc.add(rss_feed_usage)
            if search_usage:
                usage_acc.add(search_usage)

            # Checkpoint 1: Cancel prüfen nach RSS/WebSearch
            self._check_cancelled(incident_id)

            # Alle Ergebnisse zusammenführen
            all_results = rss_articles + search_results + telegram_articles + x_articles
            # Pipeline-Schritt 2: Nachrichten sammeln (fertig)
            try:
                _delivering_sources = len({a.get("source", "") for a in all_results if a.get("source")})
            except Exception:
                _delivering_sources = None
            await _pipe_done("collect", count_value=len(all_results), count_secondary=_delivering_sources)

            # Pipeline-Schritt 3: Doppeltes filtern (Start)
            await _pipe_start("dedup")

            # Duplikate entfernen (normalisierte URL + Headline-Ähnlichkeit)
            seen_urls = set()
            seen_headlines = set()
            unique_results = []
            for article in all_results:
                if not _is_duplicate(article, seen_urls, seen_headlines):
                    unique_results.append(article)

            dupes_removed = len(all_results) - len(unique_results)
            if dupes_removed > 0:
                logger.info(f"Deduplizierung: {dupes_removed} Duplikate entfernt, {len(unique_results)} verbleibend")
            await _pipe_done("dedup", count_value=dupes_removed, count_secondary=len(unique_results))

            # Relevanz-Scoring und Sortierung
            for article in unique_results:
                if "relevance_score" not in article or article["relevance_score"] == 0:
                    article["relevance_score"] = _score_relevance(article)
            unique_results.sort(key=lambda a: a.get("relevance_score", 0), reverse=True)

            source_count = len(set(a.get("source", "") for a in unique_results))
            _analyze_detail = f"Analysiert {len(unique_results)} Meldungen aus {source_count} Quellen..."
            if _pass_info:
                _analyze_detail = f"[{_pass_info['nr']}/{_pass_info['total']}] {_analyze_detail}"
            if self._ws_manager:
                await self._ws_manager.broadcast_for_incident({
                    "type": "status_update",
                    "incident_id": incident_id,
                    "data": {
                        "status": "analyzing",
                        "detail": _analyze_detail,
                        "started_at": now,
                        **_ws_extra,
                    },
                }, visibility, created_by, tenant_id)

            # --- Set-basierte DB-Deduplizierung (statt N×M Queries) ---
            # existing_db_articles_full wurde bereits oben geladen
            existing_urls = set()
            existing_headlines = set()
            for row in existing_db_articles_full:
                if row["source_url"]:
                    existing_urls.add(_normalize_url(row["source_url"]))
                if row["headline"] and len(row["headline"]) > 20:
                    norm_h = _normalize_headline(row["headline"])
                    if norm_h:
                        existing_headlines.add(norm_h)

            logger.info(f"DB-Dedup: {len(existing_urls)} URLs, {len(existing_headlines)} Headlines im Bestand")

            # --- Dedup gegen Bestand: nur neue (noch nicht gespeicherte) Kandidaten behalten ---
            new_candidates = []
            for article in unique_results:
                if article.get("source_url"):
                    norm_url = _normalize_url(article["source_url"])
                    if norm_url in existing_urls:
                        continue
                    existing_urls.add(norm_url)

                headline = article.get("headline", "")
                if headline and len(headline) > 20:
                    norm_h = _normalize_headline(headline)
                    if norm_h and norm_h in existing_headlines:
                        continue
                    if norm_h:
                        existing_headlines.add(norm_h)

                new_candidates.append(article)

            # Pipeline-Schritt 4: Relevanz bewerten (Start)
            await _pipe_start("relevance")
            _candidates_before_topic = len(new_candidates)

            # --- Pre-Topic-Übersetzung: fremdsprachige Headlines ins Englische ---
            # Damit der nachgelagerte Topic-Filter (Haiku) auch CJK/Arabisch/
            # Hebräisch/Kyrillisch-Headlines fair beurteilen kann statt sie aus
            # Sicherheit zu verwerfen.
            if new_candidates:
                try:
                    from agents.translator import translate_headlines_for_topic_filter
                    _pt_count, _pt_usage = await translate_headlines_for_topic_filter(new_candidates)
                    if _pt_usage:
                        usage_acc.add(_pt_usage)
                    if _pt_count:
                        logger.info(
                            f"Pre-Topic-Translate: {_pt_count} fremdsprachige Headlines übersetzt"
                        )
                except Exception as e:
                    logger.warning(
                        f"Pre-Topic-Translate fehlgeschlagen (Pipeline laeuft weiter): {e}"
                    )

            # --- Semantischer Topic-Filter (Haiku) ---
            # Wirft Artikel raus, die zwar Keyword-Treffer hatten, aber das Kernthema
            # der Lage nicht inhaltlich behandeln. Bei Fehler Fallback auf alle Kandidaten.
            if new_candidates:
                _tf_agent = AnalyzerAgent()
                new_candidates, _tf_usage = await _tf_agent.filter_relevant_articles(
                    title, description, new_candidates,
                )
                if _tf_usage:
                    usage_acc.add(_tf_usage)
            await _pipe_done("relevance", count_value=len(new_candidates), count_secondary=_candidates_before_topic)

            # --- Neue (thematisch gefilterte) Artikel speichern und für Analyse tracken ---
            new_count = 0
            new_articles_for_analysis = []
            for article in new_candidates:
                # headline_en / content_en: zuerst die vollwertige Übersetzung
                # vom Translator (wenn TRANSLATOR_ENABLED), sonst die für den
                # Topic-Filter angefertigte Mini-Übersetzung wiederverwenden.
                # Ohne diesen Fallback würden fremdsprachige Artikel zwar
                # gefiltert, aber ohne englische Headline in der DB landen und
                # später im Frontend bzw. im Summary-LLM unlesbar bleiben.
                headline_en = article.get("headline_en") or article.get("headline_en_for_topic")
                content_en = article.get("content_en") or article.get("content_en_for_topic")
                cursor = await db.execute(
                    """INSERT INTO articles (incident_id, headline, headline_de, headline_en, source,
                       source_url, content_original, content_de, content_en, language, published_at, tenant_id)
                       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                    (
                        incident_id,
                        article.get("headline", ""),
                        article.get("headline_de"),
                        headline_en,
                        article.get("source", "Unbekannt"),
                        article.get("source_url"),
                        article.get("content_original"),
                        article.get("content_de"),
                        content_en,
                        article.get("language", "de"),
                        article.get("published_at"),
                        tenant_id,
                    ),
                )
                new_count += 1
                article_with_id = dict(article)
                article_with_id["id"] = cursor.lastrowid
                new_articles_for_analysis.append(article_with_id)

            await db.commit()

            # Geoparsing: Orte aus neuen Artikeln extrahieren und speichern
            # Foren-Quellen (media_type='forum') ausschliessen: 5ch/Hatena/Note-Posts haben
            # keinen eigenen, fuer das Lagebild interessanten geographischen Bezug; spart Haiku-Calls.
            articles_for_geoparsing = [
                a for a in new_articles_for_analysis
                if (a.get("media_type") or "").lower() != "forum"
            ]
            if new_articles_for_analysis and not articles_for_geoparsing:
                logger.info(
                    "Geoparsing uebersprungen: alle %d neuen Artikel sind Forum-Quellen",
                    len(new_articles_for_analysis),
                )
            if articles_for_geoparsing:
                # Pipeline-Schritt 5: Orte erkennen (Start)
                await _pipe_start("geoparsing")
                try:
                    from agents.geoparsing import geoparse_articles
                    incident_context = f"{title} - {description}"
                    logger.info(f"Geoparsing fuer {len(articles_for_geoparsing)} neue Artikel (Foren ausgeschlossen)...")
                    geo_results, category_labels = await geoparse_articles(articles_for_geoparsing, incident_context)
                    geo_count = 0
                    for art_id, locations in geo_results.items():
                        for loc in locations:
                            await db.execute(
                                """INSERT INTO article_locations
                                   (article_id, incident_id, location_name, location_name_normalized,
                                    country_code, latitude, longitude, confidence, source_text, tenant_id, category)
                                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                                (art_id, incident_id, loc["location_name"], loc["location_name_normalized"],
                                 loc["country_code"], loc["lat"], loc["lon"], loc["confidence"],
                                 loc.get("source_text", ""), tenant_id, loc.get("category", "mentioned")),
                            )
                            geo_count += 1
                    if geo_count > 0:
                        await db.commit()
                        logger.info(f"Geoparsing: {geo_count} Orte aus {len(geo_results)} Artikeln gespeichert")
                    # Category-Labels in Incident speichern (nur wenn neu generiert)
                    if category_labels:
                        import json as _json
                        await db.execute(
                            "UPDATE incidents SET category_labels = ? WHERE id = ? AND category_labels IS NULL",
                            (_json.dumps(category_labels, ensure_ascii=False), incident_id),
                        )
                        await db.commit()
                        logger.info(f"Category-Labels gespeichert fuer Incident {incident_id}: {category_labels}")
                    await _pipe_done("geoparsing", count_value=geo_count, count_secondary=len(geo_results) if geo_results else 0)
                except Exception as e:
                    logger.warning(f"Geoparsing fehlgeschlagen (Pipeline laeuft weiter): {e}")
                    await _pipe_done("geoparsing", count_value=0, count_secondary=0)
            else:
                await _pipe_skip("geoparsing")

            # Quellen-Statistiken aktualisieren
            if new_count > 0:
                try:
                    from database import refresh_source_counts
                    await refresh_source_counts(db)
                except Exception as e:
                    logger.warning(f"Quellen-Statistiken konnten nicht aktualisiert werden: {e}")

            # Schritt 3+4: Analyse und Faktencheck PARALLEL
            # collect_only (Studio-Baustein "Sammeln"): ueberspringt Analyse/Faktencheck.
            if (new_count > 0 or not previous_summary) and not collect_only:
                is_first_summary = not previous_summary

                # Snapshot des alten Lagebilds sichern BEVOR parallele Verarbeitung startet
                if previous_summary:
                    cursor = await db.execute(
                        "SELECT COUNT(*) as cnt FROM articles WHERE incident_id = ?",
                        (incident_id,),
                    )
                    snap_articles = (await cursor.fetchone())["cnt"]
                    cursor = await db.execute(
                        "SELECT COUNT(*) as cnt FROM fact_checks WHERE incident_id = ?",
                        (incident_id,),
                    )
                    snap_fcs = (await cursor.fetchone())["cnt"]
                    await db.execute(
                        """INSERT INTO incident_snapshots
                           (incident_id, summary, sources_json,
                            article_count, fact_check_count, refresh_log_id, created_at, tenant_id)
                           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                        (incident_id, previous_summary, previous_sources_json,
                         snap_articles, snap_fcs, log_id, now, tenant_id),
                    )
                    await db.commit()

                # Bestehende Fakten und alle Artikel vorladen (für parallele Tasks)
                cursor = await db.execute(
                    "SELECT id, claim, status, sources_count, evidence FROM fact_checks WHERE incident_id = ?",
                    (incident_id,),
                )
                existing_facts = [dict(row) for row in await cursor.fetchall()]

                # Alle Artikel vorladen für Erstanalyse/Erstcheck
                all_articles_preloaded = None
                if not previous_summary or new_count == 0 or not existing_facts:
                    cursor = await db.execute(
                        # JOIN auf sources, damit media_type pro Artikel verfuegbar ist
                        # (Faktencheck schliesst Foren-Quellen aus, das Stimmungs-Modul nimmt
                        # nur diese). Bei Quellen ohne Match in sources bleibt media_type NULL.
                        "SELECT a.*, s.media_type AS media_type FROM articles a "
                        "LEFT JOIN sources s ON s.name = a.source "
                        "WHERE a.incident_id = ? ORDER BY a.collected_at DESC",
                        (incident_id,),
                    )
                    all_articles_preloaded = [dict(row) for row in await cursor.fetchall()]

                _parallel_detail = "Analyse und Faktencheck laufen parallel..."
                if _pass_info:
                    _parallel_detail = f"[{_pass_info['nr']}/{_pass_info['total']}] {_parallel_detail}"
                if self._ws_manager:
                    await self._ws_manager.broadcast_for_incident({
                        "type": "status_update",
                        "incident_id": incident_id,
                        "data": {"status": "analyzing", "detail": _parallel_detail, "started_at": now_utc, **_ws_extra},
                    }, visibility, created_by, tenant_id)

                # Quelleneinordnung (Bias) an Artikel anhaengen
                try:
                    cursor = await db.execute(
                        "SELECT name, domain, bias FROM sources WHERE bias IS NOT NULL"
                    )
                    _bias_rows = await cursor.fetchall()
                    _bias_by_domain = {}
                    _bias_by_name = {}
                    for br in _bias_rows:
                        brd = dict(br)
                        if brd.get("domain"):
                            _bias_by_domain[brd["domain"].lower()] = brd["bias"]
                        if brd.get("name"):
                            _bias_by_name[brd["name"].lower()] = brd["bias"]

                    def _enrich_bias(articles_list):
                        if not articles_list:
                            return
                        for art in articles_list:
                            if art.get("source_bias"):
                                continue
                            src = (art.get("source") or "").lower()
                            url = (art.get("source_url") or "").lower()
                            # Match by name
                            bias = _bias_by_name.get(src)
                            if not bias:
                                # Match by domain in URL
                                for dom, b in _bias_by_domain.items():
                                    if dom and dom in url:
                                        bias = b
                                        break
                            if bias:
                                art["source_bias"] = bias

                    _enrich_bias(new_articles_for_analysis)
                    _enrich_bias(all_articles_preloaded)
                except Exception as e:
                    logger.warning("Bias-Anreicherung fehlgeschlagen (Pipeline laeuft weiter): %s", e)

                # --- Analyse-Task (wird nach _do_factcheck mit fact_context_block aufgerufen) ---
                async def _do_analysis(fact_context_block: str = ""):
                    analyzer = AnalyzerAgent()
                    if previous_summary and new_count > 0:
                        logger.info(f"Inkrementelle Analyse: {new_count} neue Artikel zum bestehenden Lagebild")
                        return await analyzer.analyze_incremental(
                            title, description, new_articles_for_analysis,
                            previous_summary, previous_sources_json, incident_type,
                            fact_context_block=fact_context_block,
                            output_language=output_language,
                        )
                    else:
                        logger.info("Erstanalyse: Alle Artikel werden analysiert")
                        return await analyzer.analyze(
                            title, description, all_articles_preloaded, incident_type,
                            fact_context_block=fact_context_block,
                            output_language=output_language,
                        )

                # --- Faktencheck-Task ---
                async def _do_factcheck():
                    factchecker = FactCheckerAgent()
                    if existing_facts and new_count > 0:
                        if len(existing_facts) >= TWOPHASE_MIN_FACTS:
                            logger.info(
                                f"Zwei-Phasen-Faktencheck: {new_count} neue Artikel, "
                                f"{len(existing_facts)} bestehende Fakten"
                            )
                            return await factchecker.check_incremental_twophase(
                                title, new_articles_for_analysis, existing_facts, incident_type,
                                output_language=output_language,
                            )
                        else:
                            logger.info(
                                f"Inkrementeller Faktencheck: {new_count} neue Artikel, "
                                f"{len(existing_facts)} bestehende Fakten"
                            )
                            return await factchecker.check_incremental(
                                title, new_articles_for_analysis, existing_facts, incident_type,
                                output_language=output_language,
                            )
                    else:
                        # Alle Artikel laden falls nicht vorab geladen (Henne-Ei-Problem:
                        # Summary existiert aber noch keine Factchecks)
                        articles_for_check = all_articles_preloaded
                        if not articles_for_check:
                            cursor = await db.execute(
                                "SELECT * FROM articles WHERE incident_id = ? ORDER BY collected_at DESC",
                                (incident_id,),
                            )
                            articles_for_check = [dict(row) for row in await cursor.fetchall()]
                        return await factchecker.check(title, articles_for_check, incident_type, output_language=output_language)

                # Pipeline-Schritt 6: Faktencheck zuerst (sequenziell). Liefert den
                # Faktenkontext fuer das Lagebild, damit dieses auf geprueftem Stand
                # schreibt und Unklarheiten explizit benennt. Variante 1: bei
                # Faktencheck-Fehler faellt das Lagebild auf den alten Pfad ohne
                # Faktenkontext zurueck (Refresh bricht NICHT ab).
                await _pipe_start("factcheck")
                factcheck_result: tuple = ([], None)
                fact_context_block = ""
                factcheck_failed_reason: str | None = None
                try:
                    factcheck_result = await _do_factcheck()
                except Exception as fc_err:
                    factcheck_failed_reason = str(fc_err)
                    logger.warning(
                        "Faktencheck fehlgeschlagen, Lagebild laeuft ohne Faktenkontext: %s",
                        fc_err, exc_info=True,
                    )

                fact_checks, fc_usage = factcheck_result if factcheck_result else ([], None)

                # Pipeline-Schritt 6 done direkt nach dem Aufruf — die finale
                # DB-Persistierung passiert weiter unten, aber fuer die UI ist
                # der Faktencheck-Aufruf hier abgeschlossen. Der count_value
                # ist eine Schaetzung (echte Zahl steht spaeter in der DB).
                _fc_estimated_new = max(0, len(fact_checks or []) - len(existing_facts or []))
                await _pipe_done(
                    "factcheck",
                    count_value=_fc_estimated_new,
                    count_secondary=len(fact_checks) if fact_checks else 0,
                )

                # Faktenkontext fuer das Lagebild bauen.
                try:
                    from agents.analyzer import build_fact_context_block as _build_fc_ctx
                    fact_context_block = _build_fc_ctx(
                        existing_facts or [], fact_checks or [], incident_type,
                    )
                    if fact_context_block:
                        logger.info(
                            "Faktenkontext fuer Lagebild: %d Zeichen, basierend auf %d alten + %d neuen Fakten",
                            len(fact_context_block), len(existing_facts or []), len(fact_checks or []),
                        )
                except Exception as ctx_err:
                    logger.warning("build_fact_context_block fehlgeschlagen: %s", ctx_err, exc_info=True)
                    fact_context_block = ""

                # Pipeline-Schritt 6b: Öffentliche Stimmung aus Foren-Quellen
                # (nur Artikel mit media_type='forum'). Eigene Kachel, kein Faktencheck.
                # Wird vor dem Lagebild-Schritt ausgefuehrt, damit das Lagebild bei
                # Bedarf darauf verweisen kann (z.B. Demo-Lagen mit Bezug zur Stimmung).
                try:
                    # Bestand aller Foren-Artikel der Lage laden (inkl. media_type via JOIN)
                    cursor_fm = await db.execute(
                        "SELECT a.*, s.media_type AS media_type FROM articles a "
                        "LEFT JOIN sources s ON s.name = a.source "
                        "WHERE a.incident_id = ?",
                        (incident_id,),
                    )
                    all_articles_with_mt = [dict(r) for r in await cursor_fm.fetchall()]
                    forum_articles_in_db = [
                        a for a in all_articles_with_mt
                        if (a.get("media_type") or "").lower() == "forum"
                    ]
                    # Aus dem aktuellen Refresh-Lauf zusaetzliche Foren-Artikel ergaenzen
                    # (haben media_type aus feed_config, sind aber evtl. noch nicht in DB,
                    # wenn die Persistierung anders laeuft — Robustheit).
                    for art in new_articles_for_analysis:
                        if (art.get("media_type") or "").lower() != "forum":
                            continue
                        # Duplikate vermeiden ueber source_url
                        if any(a.get("source_url") == art.get("source_url") for a in forum_articles_in_db):
                            continue
                        forum_articles_in_db.append(art)

                    if forum_articles_in_db:
                        await _pipe_start("public_mood")
                        try:
                            mood_agent = AnalyzerAgent()
                            # 1. Moderationspass: Hassrede/PII/NSFW vorab filtern.
                            moderated_articles, mod_usage = await mood_agent.moderate_forum_articles(
                                forum_articles_in_db,
                            )
                            if mod_usage:
                                usage_acc.add(mod_usage)
                            # 2. Stimmungs-Zusammenfassung aus gefilterten Beitraegen.
                            mood_text, mood_usage = await mood_agent.generate_public_mood(
                                title, description, moderated_articles,
                                output_language=output_language,
                            )
                            if mood_usage:
                                usage_acc.add(mood_usage)
                            if mood_text:
                                await db.execute(
                                    "UPDATE incidents SET public_mood = ?, public_mood_updated_at = ? WHERE id = ?",
                                    (mood_text, now, incident_id),
                                )
                                await db.commit()
                                logger.info(
                                    "Public-Mood gespeichert fuer Incident %d (%d -> %d Foren-Artikel nach Moderation)",
                                    incident_id, len(forum_articles_in_db), len(moderated_articles),
                                )
                            await _pipe_done(
                                "public_mood",
                                count_value=len(moderated_articles),
                                count_secondary=(1 if mood_text else 0),
                            )
                        except Exception as mood_err:
                            logger.warning("Public-Mood fehlgeschlagen: %s", mood_err, exc_info=True)
                            await _pipe_done("public_mood", count_value=0, count_secondary=0)
                    else:
                        await _pipe_skip("public_mood")
                except Exception as mood_outer_err:
                    logger.warning("Public-Mood-Block uebersprungen: %s", mood_outer_err)
                    try:
                        await _pipe_skip("public_mood")
                    except Exception:
                        pass

                # Pipeline-Schritt 7: Lagebild verfassen (jetzt mit Faktenkontext)
                await _pipe_start("summary")
                logger.info(
                    "Starte Lagebild (sequenziell nach Faktencheck%s)",
                    " — OHNE Faktenkontext (Fallback)" if factcheck_failed_reason else "",
                )
                analysis_result = await _do_analysis(fact_context_block)

                analysis, analysis_usage = analysis_result
                await _pipe_done("summary", count_value=None, count_secondary=None)

                # --- Analyse-Ergebnisse verarbeiten ---
                if analysis_usage:
                    usage_acc.add(analysis_usage)

                if analysis:
                    sources = analysis.get("sources", [])
                    new_summary = analysis.get("summary", "") or previous_summary

                    # Validierung: Fehlende Quellennummern im Summary erkennen und reparieren
                    if sources and new_summary:
                        import re as _re
                        # Auch alphanumerische Refs wie [389a] erkennen
                        referenced_raw = set(_re.findall(r'\[(\d+[a-z]?)\]', new_summary))
                        referenced_nrs = set()
                        for r in referenced_raw:
                            try:
                                referenced_nrs.add(int(r))
                            except ValueError:
                                referenced_nrs.add(r)  # Keep alphanumeric as string
                        defined_nrs = set()
                        for s in sources:
                            nr = s.get("nr", 0)
                            if isinstance(nr, int):
                                defined_nrs.add(nr)
                            elif isinstance(nr, str):
                                try:
                                    defined_nrs.add(int(nr))
                                except ValueError:
                                    defined_nrs.add(nr)  # Keep alphanumeric like '389a'
                        missing_nrs = sorted(referenced_nrs - defined_nrs)
                        if missing_nrs:
                            truly_missing = []
                            for nr in missing_nrs:
                                # Buchstaben-Suffix (z.B. "22b") -> Basisnummer (22) aufloesen
                                if isinstance(nr, str) and _re.match(r"^\d+[a-z]$", nr):
                                    base_nr = int(nr[:-1])
                                    if base_nr in defined_nrs:
                                        new_summary = new_summary.replace(f"[{nr}]", f"[{base_nr}]")
                                        logger.info(
                                            "Incident %d: Suffix-Ref [%s] auf Basisquelle [%d] aufgeloest",
                                            incident_id, nr, base_nr
                                        )
                                        continue
                                truly_missing.append(nr)
                            if truly_missing:
                                logger.warning(
                                    "Incident %d: %d Quellennummern im Summary ohne Eintrag in sources: %s",
                                    incident_id, len(truly_missing), truly_missing[:20]
                                )
                                for nr in truly_missing:
                                    sources.append({"nr": nr, "name": "Quelle", "url": ""})
                                    logger.info("Platzhalter fuer fehlende Quelle [%s] eingefuegt", nr)
                                sources.sort(key=lambda s: int(s.get("nr", 0)) if isinstance(s.get("nr"), int) or (isinstance(s.get("nr"), str) and str(s.get("nr", "")).isdigit()) else 9999)

                    # Sicherstellen dass alle nr-Werte Integer sind (Claude liefert manchmal Strings)
                    if sources:
                        for s in sources:
                            nr = s.get("nr")
                            if isinstance(nr, str):
                                try:
                                    s["nr"] = int(nr)
                                except ValueError:
                                    pass
                    sources_json = json.dumps(sources, ensure_ascii=False) if sources else previous_sources_json

                    # summary_at haelt fest, WANN das Lagebild entstand (Studio-Freshness).
                    # JETZT stempeln, nicht 'now' vom Lauf-Beginn: sonst saehen die in diesem
                    # Lauf gesammelten Artikel neuer aus als der Bericht, der sie schon enthaelt.
                    summary_now = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')
                    await db.execute(
                        "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, "
                        "updated_at = ?, summary_at = ? WHERE id = ?",
                        (new_summary, sources_json, summary_now, summary_now, incident_id),
                    )

                    # Beim ersten Refresh: Snapshot des neuen Lagebilds erstellen
                    if is_first_summary and new_summary:
                        cursor = await db.execute(
                            "SELECT COUNT(*) as cnt FROM articles WHERE incident_id = ?",
                            (incident_id,),
                        )
                        snap_articles = (await cursor.fetchone())["cnt"]
                        cursor = await db.execute(
                            "SELECT COUNT(*) as cnt FROM fact_checks WHERE incident_id = ?",
                            (incident_id,),
                        )
                        snap_fcs = (await cursor.fetchone())["cnt"]
                        await db.execute(
                            """INSERT INTO incident_snapshots
                               (incident_id, summary, sources_json,
                                article_count, fact_check_count, refresh_log_id, created_at, tenant_id)
                               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                            (incident_id, new_summary, sources_json,
                             snap_articles, snap_fcs, log_id, now, tenant_id),
                        )

                    # Translations werden vom dedizierten Translator-Agent unten
                    # erzeugt (frueher inline im Analyzer-Output, das war token-
                    # instabil und schaetzte regelmaessig content_de aus).

                    await db.commit()

                # Cancel-Check nach paralleler Verarbeitung
                self._check_cancelled(incident_id)

                # --- Translator (Haiku) fuer fremdsprachige Artikel ohne DE-Texte ---
                # Idempotent: nur Artikel ohne headline_de/content_de werden geholt.
                # Lauft nach der Analyse (Lagebild ist schon committed) und vor QC
                # (damit normalize_umlaut_articles auch die frischen DE-Texte fasst).
                _translate_step_started = False
                try:
                    tr_cursor = await db.execute(
                        """SELECT id, headline, content_original, language
                           FROM articles
                           WHERE incident_id = ?
                             AND language IS NOT NULL AND LOWER(language) != 'de'
                             AND (headline_de IS NULL OR headline_de = ''
                                  OR content_de IS NULL OR content_de = '')""",
                        (incident_id,),
                    )
                    pending_translations = [dict(r) for r in await tr_cursor.fetchall()]
                    if pending_translations and translator_enabled:
                        # Pipeline-Schritt 9: Artikel uebersetzen (nur sichtbar wenn was zu uebersetzen)
                        await _pipe_start("translate")
                        _translate_step_started = True
                        logger.info(
                            "Translator fuer Incident %d: %d Artikel ohne DE-Uebersetzung",
                            incident_id, len(pending_translations),
                        )
                        from agents.translator import translate_articles
                        from services.post_refresh_qc import normalize_german_umlauts as _norm_de2
                        translations = await translate_articles(
                            pending_translations,
                            output_lang=output_language_iso,
                            usage_accumulator=usage_acc,
                            enabled=translator_enabled,
                        )
                        for t in translations:
                            hd = t.get("headline_de")
                            cd = t.get("content_de")
                            if hd:
                                hd, _ = _norm_de2(hd)
                            if cd:
                                cd, _ = _norm_de2(cd)
                            if hd or cd:
                                await db.execute(
                                    "UPDATE articles SET headline_de = COALESCE(?, headline_de), "
                                    "content_de = COALESCE(?, content_de) WHERE id = ? AND incident_id = ?",
                                    (hd, cd, t["id"], incident_id),
                                )
                        await db.commit()
                        logger.info(
                            "Translator fuer Incident %d: %d/%d Artikel uebersetzt",
                            incident_id, len(translations), len(pending_translations),
                        )
                        await _pipe_done("translate", count_value=len(translations), count_secondary=len(pending_translations))
                except Exception as e:
                    logger.error("Translator-Fehler fuer Incident %d: %s", incident_id, e, exc_info=True)
                    if _translate_step_started:
                        await _pipe_done("translate", count_value=0, count_secondary=0)
                    # Refresh trotz Translator-Fehler weiterlaufen lassen

                # --- Neueste Entwicklungen (nur Live-Monitoring / adhoc) ---
                # Basis ist jetzt das frisch generierte Lagebild (autoritativ, thematisch sauber).
                # Zeitstempel und Quellen kommen aus den jüngsten belegenden Artikeln.
                dev_summary_source = (locals().get("new_summary") or previous_summary or "").strip()
                if incident_type == "adhoc" and dev_summary_source:
                    try:
                        # Top-60 neueste Artikel mit Publikationsdatum als Beleg-Pool.
                        dev_cursor = await db.execute(
                            """SELECT id, headline, headline_de, source, source_url, published_at
                               FROM articles
                               WHERE incident_id = ? AND published_at IS NOT NULL
                               ORDER BY published_at DESC LIMIT 60""",
                            (incident_id,),
                        )
                        dev_articles = [dict(row) for row in await dev_cursor.fetchall()]

                        dev_analyzer = AnalyzerAgent()
                        dev_text, dev_usage = await dev_analyzer.generate_latest_developments(
                            title, description, dev_summary_source, dev_articles, previous_developments,
                            output_language=output_language,
                        )
                        if dev_usage:
                            usage_acc.add(dev_usage)
                        if dev_text is not None:
                            await db.execute(
                                "UPDATE incidents SET latest_developments = ? WHERE id = ?",
                                (dev_text, incident_id),
                            )
                            await db.commit()
                            previous_developments = dev_text
                    except Exception as e:
                        logger.warning(f"Latest-Developments-Generator fehlgeschlagen: {e}")

                # Cancel-Check nach Analyse+Faktencheck
                self._check_cancelled(incident_id)

                # --- Faktencheck-Ergebnisse verarbeiten ---
                # Pre-Dedup: Duplikate aus LLM-Antwort entfernen
                fact_checks = deduplicate_new_facts(fact_checks)

                if fc_usage:
                    usage_acc.add(fc_usage)

                # Prüfen ob dies der erste Refresh ist (keine vorherigen Faktenchecks)
                is_first_refresh = len(existing_facts) == 0

                # Notification-Summary sammeln
                confirmed_count = 0
                contradicted_count = 0
                status_changes = []

                # --- Schutz gegen Massen-Downgrades ---
                # Wenn >50% der established/confirmed Fakten auf unverified/unconfirmed
                # herabgestuft wuerden, verwerfe die FC-Ergebnisse komplett.
                established_ids = {ef["id"] for ef in existing_facts if ef.get("status") in ("established", "confirmed")}
                if established_ids and fact_checks:
                    _downgrade_count = 0
                    _remaining_tmp = list(existing_facts)
                    for _fc in fact_checks:
                        _matched = find_matching_claim(_fc.get("claim", ""), _remaining_tmp)
                        if _matched and _matched["id"] in established_ids:
                            _new_st = _fc.get("status", "developing")
                            if _new_st in ("unverified", "unconfirmed", "developing"):
                                _downgrade_count += 1
                            _remaining_tmp = [ef for ef in _remaining_tmp if ef["id"] != _matched["id"]]
                    _downgrade_ratio = _downgrade_count / len(established_ids) if established_ids else 0
                    if _downgrade_ratio > 0.5:
                        logger.warning(
                            f"Faktencheck-Ergebnisse verworfen: {_downgrade_count}/{len(established_ids)} "
                            f"established Fakten wuerden herabgestuft ({_downgrade_ratio:.0%}). "
                            f"Bestehende Fakten bleiben unveraendert."
                        )
                        fact_checks = []

                # Mutable Kopie für Fuzzy-Matching
                remaining_existing = list(existing_facts)

                for fc in fact_checks:
                    new_claim = fc.get("claim", "")
                    new_status = fc.get("status", "developing")

                    # Fuzzy-Matching gegen bestehende Claims
                    matched = find_matching_claim(new_claim, remaining_existing)

                    if matched:
                        old_status = matched.get("status")
                        # status_history aktualisieren bei Statusaenderung
                        history_update = ""
                        if old_status and old_status != new_status:
                            import json as _json
                            cursor_hist = await db.execute(
                                "SELECT status_history FROM fact_checks WHERE id = ?",
                                (matched["id"],),
                            )
                            hist_row = await cursor_hist.fetchone()
                            try:
                                history = _json.loads(hist_row[0] or "[]") if hist_row else []
                            except (ValueError, TypeError):
                                history = []
                            history.append({"status": new_status, "at": now})
                            history_update = _json.dumps(history)
                        # Evidence: Alte URLs beibehalten wenn neue keine hat
                        new_evidence = fc.get("evidence") or ""
                        import re as _re
                        if not _re.search(r"https?://", new_evidence) and matched.get("evidence"):
                            old_evidence = matched["evidence"] or ""
                            if _re.search(r"https?://", old_evidence):
                                bracket_match = _re.search(r"\[(?:Quellen|Weitere Quellen|Ursprungsquellen):.*?\]", old_evidence)
                                if bracket_match:
                                    new_evidence = new_evidence.rstrip(". ") + ". " + bracket_match.group()
                                else:
                                    new_evidence = old_evidence

                        await db.execute(
                            "UPDATE fact_checks SET claim = ?, status = ?, sources_count = ?, evidence = ?, is_notification = ?, checked_at = ?"
                            + (", status_history = ?" if history_update else "")
                            + " WHERE id = ?",
                            (new_claim, new_status, fc.get("sources_count", 0), new_evidence, fc.get("is_notification", 0), now)
                            + ((history_update,) if history_update else ())
                            + (matched["id"],),
                        )
                        # Aus der Liste entfernen damit nicht doppelt gematcht wird
                        remaining_existing = [ef for ef in remaining_existing if ef["id"] != matched["id"]]

                        # Status-Änderung tracken
                        if not is_first_refresh and old_status and old_status != new_status:
                            status_changes.append({
                                "claim": new_claim,
                                "old_status": old_status,
                                "new_status": new_status,
                            })
                    else:
                        import json as _json
                        initial_history = _json.dumps([{"status": new_status, "at": now}])
                        await db.execute(
                            """INSERT INTO fact_checks (incident_id, claim, status, sources_count, evidence, is_notification, tenant_id, status_history)
                               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                            (incident_id, new_claim, new_status, fc.get("sources_count", 0), fc.get("evidence"), fc.get("is_notification", 0), tenant_id, initial_history),
                        )

                    # Status-Statistik sammeln
                    if new_status in ("confirmed", "established"):
                        confirmed_count += 1
                    elif new_status in ("contradicted", "disputed", "false"):
                        contradicted_count += 1

                await db.commit()

                # Pipeline-Schritt 7 (Fakten pruefen) wurde bereits frueher als done
                # markiert (siehe weiter oben — direkt nach dem _do_factcheck-Aufruf,
                # bevor das Lagebild generiert wurde). Hier nur noch die DB-
                # Persistierung der Fakten, ohne den Step erneut zu schliessen.

                # Pipeline-Schritt 8: Qualitätscheck (Start, ohne Zahlen)
                await _pipe_start("qc")

                # Post-Refresh Quality Check: Duplikate und Karten-Kategorien pruefen
                try:
                    from services.post_refresh_qc import run_post_refresh_qc
                    qc_result = await run_post_refresh_qc(db, incident_id)
                    if qc_result.get("facts_removed", 0) > 0 or qc_result.get("locations_fixed", 0) > 0:
                        logger.info(
                            f"QC: {qc_result['facts_removed']} Duplikate, "
                            f"{qc_result['locations_fixed']} Location-Fixes"
                        )
                except Exception as qc_err:
                    logger.warning(f"Post-Refresh QC fehlgeschlagen: {qc_err}")
                await _pipe_done("qc", count_value=None, count_secondary=None)

                # Pipeline-Schritt 9: Benachrichtigen (Start)
                await _pipe_start("notify")
                _notify_count = 0

                # Gebündelte Notification senden (nicht beim ersten Refresh)
                if not is_first_refresh:
                    if self._ws_manager:
                        await self._ws_manager.broadcast_for_incident({
                            "type": "refresh_summary",
                            "incident_id": incident_id,
                            "data": {
                                "new_articles": new_count,
                                "confirmed_count": confirmed_count,
                                "contradicted_count": contradicted_count,
                                "status_changes": status_changes,
                                "is_first_refresh": False,
                                "incident_title": title,
                            },
                        }, visibility, created_by, tenant_id)

                    # DB-Notifications erzeugen (Texte org-sprach-relativ)
                    is_en = output_language_iso == "en"
                    parts = []
                    if is_en:
                        if new_count > 0:
                            parts.append(f"{new_count} new article{'s' if new_count != 1 else ''}")
                        if confirmed_count > 0:
                            parts.append(f"{confirmed_count} confirmed")
                        if contradicted_count > 0:
                            parts.append(f"{contradicted_count} contradicted")
                        summary_text = ", ".join(parts) if parts else "No new developments"
                        research_prefix = "Research"
                        new_articles_msg = f"{new_count} new article{'s' if new_count != 1 else ''} found"
                    else:
                        if new_count > 0:
                            parts.append(f"{new_count} neue Meldung{'en' if new_count != 1 else ''}")
                        if confirmed_count > 0:
                            parts.append(f"{confirmed_count} bestätigt")
                        if contradicted_count > 0:
                            parts.append(f"{contradicted_count} widersprochen")
                        summary_text = ", ".join(parts) if parts else "Keine neuen Entwicklungen"
                        research_prefix = "Recherche"
                        new_articles_msg = f"{new_count} neue Meldung{'en' if new_count != 1 else ''} gefunden"

                    db_notifications = [{
                        "type": "refresh_summary",
                        "title": title,
                        "text": f"{research_prefix}: {summary_text}",
                        "icon": "warning" if contradicted_count > 0 else "success",
                    }]
                    if new_count > 0:
                        db_notifications.append({
                            "type": "new_articles",
                            "title": title,
                            "text": new_articles_msg,
                            "icon": "info",
                        })
                    for sc in status_changes:
                        db_notifications.append({
                            "type": "status_change",
                            "title": title,
                            "text": f"{sc['claim']}: {sc['old_status']} \u2192 {sc['new_status']}",
                            "icon": "error" if sc["new_status"] in ("contradicted", "disputed", "false") else "success",
                        })

                    if created_by:
                        await _create_notifications_for_incident(
                            db, incident_id, visibility, created_by, tenant_id, db_notifications
                        )
                        # E-Mail-Benachrichtigungen versenden
                        await _send_email_notifications_for_incident(
                            db, incident_id, title, visibility, created_by, tenant_id, db_notifications,
                            incident_type=incident_type,
                        )
                        _notify_count = len(db_notifications)

                # Pipeline-Schritt 9: Benachrichtigen (fertig)
                await _pipe_done("notify", count_value=_notify_count, count_secondary=None)

            # Falls Analyse-Block übersprungen wurde (kein neuer Artikel und Summary existiert),
            # die noch offenen Pipeline-Schritte als übersprungen markieren.
            for _skipped_key in ("summary", "factcheck", "qc", "notify"):
                if _skipped_key in _step_ids or _skipped_key not in {"summary", "factcheck", "qc", "notify"}:
                    pass
            # Saubere Variante: alle noch offenen Steps am Ende skippen
            for _open_key in list(_step_ids.keys()):
                await _pipe_skip(_open_key)
            # Auch Steps die nie gestartet wurden (bei übersprungenem Outer-If)
            _started_keys = set()
            try:
                _check_cursor = await db.execute(
                    "SELECT step_key FROM refresh_pipeline_steps WHERE refresh_log_id = ? AND pass_number = ?",
                    (log_id, _pass_nr),
                )
                _started_keys = {row[0] for row in await _check_cursor.fetchall()}
            except Exception:
                pass
            for _missing_key in ("summary", "factcheck", "qc", "notify"):
                if _missing_key not in _started_keys:
                    await _pipe_skip(_missing_key)

            # Refresh-Log abschließen (mit Token-Statistiken). Störungen des
            # KI-Dienstes werden auch bei einem erfolgreichen Lauf vermerkt,
            # sonst sieht ein Durchlauf mit Aussetzern aus wie ein glatter.
            stoerungshinweis = bedrock_client.stoerungen_zusammenfassen() or None
            if stoerungshinweis:
                logger.info("Refresh-Protokoll Lage %s: %s", incident_id, stoerungshinweis)
            await db.execute(
                """UPDATE refresh_log SET
                    completed_at = ?, articles_found = ?, status = 'completed',
                    input_tokens = ?, output_tokens = ?,
                    cache_creation_tokens = ?, cache_read_tokens = ?,
                    total_cost_usd = ?, api_calls = ?, error_message = ?
                WHERE id = ?""",
                (datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S'), new_count,
                 usage_acc.input_tokens, usage_acc.output_tokens,
                 usage_acc.cache_creation_tokens, usage_acc.cache_read_tokens,
                 round(usage_acc.total_cost_usd, 7), usage_acc.call_count,
                 stoerungshinweis, log_id),
            )
            await db.commit()
            logger.info(
                f"Token: {usage_acc.input_tokens} in / {usage_acc.output_tokens} out / "
                f"${usage_acc.total_cost_usd:.4f} ({usage_acc.call_count} Calls)"
            )

            # Guthaben belasten + monatliche Statistik fortschreiben. Der Lagentyp
            # entscheidet ueber den Tarif, ein Research-Durchlauf kostet mehr als
            # ein Live-Refresh. Nicht mehr an total_cost_usd > 0 gebunden, weil im
            # Pauschalmodus die Aktion zaehlt und lokale Modelle 0 kosten.
            if tenant_id:
                from services.license_service import charge_usage_to_tenant
                await charge_usage_to_tenant(
                    db, tenant_id, usage_acc, source="monitor", incident_type=incident_type
                )
                await db.commit()

            # Quellen-Discovery im Background starten
            if unique_results:
                asyncio.create_task(_background_discover_sources(unique_results))

            if not _suppress_complete and self._ws_manager:
                await self._ws_manager.broadcast_for_incident({
                    "type": "refresh_complete",
                    "incident_id": incident_id,
                    "data": {"new_articles": new_count, "status": "idle"},
                }, visibility, created_by, tenant_id)

            # updated_at IMMER aktualisieren wenn Refresh lief (auch bei fehlgeschlagener Analyse)
            await db.execute(
                "UPDATE incidents SET updated_at = ? WHERE id = ?",
                (now, incident_id),
            )
            await db.commit()

            logger.info(f"Refresh für Lage {incident_id} abgeschlossen: {new_count} neue Artikel")

            # Multi-Pass-Diagnose: Pass-Ergebnis zurueck an Multi-Pass-Caller geben
            if _pass_info is not None:
                _pass_info["new_count"] = new_count
                _pass_info["parse_failed"] = search_parse_failed

            # Executive Summary im Hintergrund vorab generieren (fuer schnelleren Export)
            if new_count > 0:
                async def _pregenerate_exec_summary():
                    try:
                        from report_generator import generate_executive_summary
                        from database import get_db
                        _db = await get_db()
                        try:
                            cursor = await _db.execute(
                                "SELECT summary, executive_summary FROM incidents WHERE id = ?",
                                (incident_id,),
                            )
                            _row = await cursor.fetchone()
                            if _row and _row["summary"] and not _row["executive_summary"]:
                                es = await generate_executive_summary(_row["summary"])
                                await _db.execute(
                                    "UPDATE incidents SET executive_summary = ? WHERE id = ?",
                                    (es, incident_id),
                                )
                                await _db.commit()
                                logger.info(f"Executive Summary fuer Lage {incident_id} vorberechnet")
                        finally:
                            await _db.close()
                    except Exception as e:
                        logger.warning(f"Executive Summary Vorberechnung fehlgeschlagen: {e}")
                asyncio.create_task(_pregenerate_exec_summary())

        finally:
            await db.close()


    async def _get_incident_info(self, incident_id: int) -> tuple[str, bool]:
        """Incident-Typ und Summary-Status laden."""
        from database import get_db
        db = await get_db()
        try:
            cursor = await db.execute(
                "SELECT type, summary FROM incidents WHERE id = ?", (incident_id,)
            )
            row = await cursor.fetchone()
            if not row:
                return "adhoc", False
            return row["type"] or "adhoc", bool(row["summary"])
        finally:
            await db.close()

    async def _run_research_multi_pass(self, incident_id: int, trigger_type: str, user_id: int = None):
        """Führt automatisch 3 Recherche-Durchläufe für Research-Lagen durch.

        Durchlauf 1: Breite Erfassung (initiale 4-Phasen-Recherche)
        Durchlauf 2: Vertiefung (andere Quellen, inkrementelle Analyse)
        Durchlauf 3: Konsolidierung (letzte Lücken, Fakten-Upgrade)
        """
        total = RESEARCH_MULTI_PASS_COUNT
        pass_results = []

        for pass_nr in range(1, total + 1):
            # Cancel zwischen Durchläufen prüfen
            self._check_cancelled(incident_id)

            is_last = (pass_nr == total)
            pass_info = {
                "nr": pass_nr,
                "total": total,
                "label": RESEARCH_PASS_LABELS.get(pass_nr, f"Durchlauf {pass_nr}"),
            }

            logger.info(
                f"Research Multi-Pass {pass_nr}/{total} für Lage {incident_id}: "
                f"{pass_info['label']}"
            )

            try:
                await self._run_refresh(
                    incident_id,
                    trigger_type=trigger_type,
                    retry_count=0,
                    user_id=user_id,
                    _suppress_complete=not is_last,
                    _pass_info=pass_info,
                )
            except asyncio.CancelledError:
                logger.info(
                    f"Research Multi-Pass abgebrochen in Durchlauf {pass_nr}/{total} "
                    f"für Lage {incident_id}"
                )
                raise
            except Exception as e:
                logger.error(
                    f"Research Multi-Pass {pass_nr}/{total} fehlgeschlagen "
                    f"für Lage {incident_id}: {e}"
                )
                if is_last:
                    raise
                # Nicht-letzter Durchlauf: weiter mit nächstem, bisherige Ergebnisse bleiben
            finally:
                pass_results.append(pass_info)

        logger.info(
            f"Research Multi-Pass abgeschlossen für Lage {incident_id}: "
            f"{total} Durchläufe"
        )

        # Diagnose: Wenn ALLE Passes 0 neue Artikel hatten UND mindestens einer
        # an einem Parser-Fehler scheiterte, ist die Recherche faktisch fehlgeschlagen —
        # Claude lieferte zwar Antworten, aber kein verwertbares JSON. Sonst bliebe
        # die Lage ohne sichtbare Fehlermeldung leer (siehe staging Lage "Friedrich Merz").
        total_new = sum(p.get("new_count", 0) for p in pass_results)
        any_parse_failed = any(p.get("parse_failed") for p in pass_results)
        if total_new == 0 and any_parse_failed:
            raise RuntimeError(
                "Recherche fehlgeschlagen: Claude lieferte keine verwertbaren Quellen "
                "(JSON-Parsing schlug bei mindestens einem Durchlauf fehl). "
                "Bitte Logs prüfen und Refresh erneut starten."
            )


# Singleton-Instanz
orchestrator = AgentOrchestrator()