src/app.py

from __future__ import annotations

import os
import re
import logging
from huggingface_hub import snapshot_download
from dataclasses import dataclass
from pathlib import Path
from threading import Lock
from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, TypedDict
from uuid import uuid4

# .env laden, bevor irgendein os.getenv() ausgewertet wird.
#
# Ohne diesen Schritt ist die gesamte .env tote Konfiguration: CHROMA_COLLECTIONS
# erreicht specs_from_env() nie, die Registry fällt auf das einzelne
# CHROMA_COLLECTION zurück und das zweite Korpus (SGB V) wird nie geöffnet.
#
# override=False ist Absicht: echte Prozess-Umgebungsvariablen (Render, Docker,
# HF Spaces) haben Vorrang vor der lokalen Entwicklungsdatei.
try:
    from dotenv import load_dotenv

    _ENV_FILE = Path(__file__).resolve().parent.parent / ".env"
    load_dotenv(_ENV_FILE if _ENV_FILE.exists() else None, override=False)
except ImportError:  # python-dotenv ist optional; Deployment setzt echte Env-Vars.
    _ENV_FILE = None

from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from retriever import LegalRetriever
from corpus_amendments import (
    append_amendment_note,
    describe as describe_amendments,
    reconcile_negative_answer,
)
from corpus_boundary import append_boundary_note, detect_external_references
import amrl_austauschbarkeit
import amrl_biosimilars
import amrl_lifestyle
import amrl_otc
import amrl_substitution
import amrl_tabakentwoehnung
import amrl_verordnungsausschluss
import norm_anchors
import norm_verweise
from corpus_registry import CorpusRegistry, build_registry
from corpus_router import explain_routing, mentioned_corpora, route_question
from federated_retriever import FederatedRetriever
from llm_client_groq import (
    DEFAULT_SYSTEM_PROMPT,
    ConversationMemory,
    GroqClient,
    classify_question,
    is_meta_question,
)
from answer_composer import AnswerComposer
from orchestrator import NEGATIVE_ANSWER_RE, LegalAnswerOrchestrator, OrchestratorOptions

try:
    from langgraph.graph import StateGraph, END
except ImportError:  # LangGraph ist nur für Workflow-Visualisierung erforderlich.
    StateGraph = None  # type: ignore[assignment]
    END = "__end__"  # type: ignore[assignment]


# -----------------------------------------------------------------------------
# Konfiguration
# -----------------------------------------------------------------------------

logger = logging.getLogger(__name__)

APP_TITLE = os.getenv("APP_TITLE", "Juristischer RAG-Prototyp")
APP_VERSION = os.getenv("APP_VERSION", "1.1")

# Robuste Pfad-Ermittlung:
# Funktioniert sowohl für:
#   repo/app.py
# als auch für:
#   repo/src/app.py
APP_DIR = Path(__file__).resolve().parent

if (APP_DIR / "static").exists():
    BASE_DIR = APP_DIR
else:
    BASE_DIR = APP_DIR.parent


# ---- HF Dataset -> lokale DB nach /data ziehen (Spaces) ----

HF_DATASET_REPO = os.getenv("HF_DATASET_REPO", "AlixJabda/bav-ki-db")
DATA_DIR = Path(os.getenv("DATA_DIR", "/data"))
CHROMA_DATA_DIR = DATA_DIR / "chroma_db"

# Optional Token, wenn Dataset private ist
HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN")

# Nur einmal laden, wenn noch nicht vorhanden
try:
    if not CHROMA_DATA_DIR.exists():
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        snapshot_download(
            repo_id=HF_DATASET_REPO,
            repo_type="dataset",
            local_dir=str(DATA_DIR),
            token=HF_TOKEN,
        )
        logger.warning("HF dataset geladen nach %s", CHROMA_DATA_DIR)
except Exception as exc:
    logger.warning("HF dataset download fehlgeschlagen: %s", exc)


def _resolve_project_path(env_name: str, default_relative: str | Path) -> str:
    raw = os.getenv(env_name)
    path = Path(raw) if raw else Path(default_relative)

    if not path.is_absolute():
        path = BASE_DIR / path

    return str(path.resolve())


# Deine aktuelle Struktur:
#
# static/
# ├── index.html
# ├── js/
# │   └── script.js
# ├── styles/
# │   └── style.css
# └── berliner_apotheker_verein_cover.jpg

STATIC_DIR = _resolve_project_path("STATIC_DIR", "static")
INDEX_FILE = _resolve_project_path("INDEX_FILE", Path("static") / "index.html")


# --- PDF-Quelldokumente für die interaktive Fundstellen-Anzeige ---
#
# Suchreihenfolge:
#   1. PDF_DIR (Env, explizit)
#   2. DATA_DIR/pdfs   -> HF-Dataset-Snapshot auf Spaces (dort pdfs/ ablegen)
#   3. BASE_DIR/data/pdfs bzw. BASE_DIR/data -> im Docker-Image mitgelieferte PDFs
# Kuratierte Änderungen zu Korpora, die nur als Grundfassung plus separate
# Änderungstexte veröffentlicht werden (siehe src/corpus_amendments.py).
AMENDMENTS_FILE = Path(
    os.getenv("AMENDMENTS_FILE", str(BASE_DIR / "data" / "amabrv_aenderungen.json"))
)
# Teil B der AM-RL-Anlage VII als nachschlagbare Tabelle. Diese Liste wird nicht
# indiziert: die entscheidende Auskunft ist oft die negative, und die kann ein
# Vektor-Retrieval nicht belegen (siehe src/amrl_substitution.py).
AMRL_SUBSTITUTION_FILE = Path(
    os.getenv(
        "AMRL_SUBSTITUTION_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_vii_teil_b.json"),
    )
)
# Teil A derselben Anlage: welche Darreichungsformen desselben Wirkstoffs
# gegeneinander austauschbar sind. Ebenfalls Lookup statt Index — die Aussage
# hängt an der Gruppenzugehörigkeit, und die ist nur über die vollständige
# Tabelle bestimmbar (siehe src/amrl_austauschbarkeit.py).
AMRL_AUSTAUSCHBARKEIT_FILE = Path(
    os.getenv(
        "AMRL_AUSTAUSCHBARKEIT_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_vii_teil_a.json"),
    )
)
# Anlage VIIa: welches Biosimilar zu welchem Referenzarzneimittel zugelassen ist.
# Ebenfalls Lookup — die Zuordnung hängt an der Tabellenzeile, und ein Retrieval
# zeigte Prolia und ein Xgeva-Biosimilar nebeneinander, weil beide unter
# „Denosumab" stehen (siehe src/amrl_biosimilars.py).
AMRL_BIOSIMILARS_FILE = Path(
    os.getenv(
        "AMRL_BIOSIMILARS_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_viia.json"),
    )
)
# Anlage I: die OTC-Ausnahmen vom Verordnungsausschluss des § 34 Absatz 1 SGB V.
# Lookup wie die übrigen — und hier trägt die Negativauskunft ausdrücklich, weil
# § 12 Absatz 10 AM-RL abschließend regelt (siehe src/amrl_otc.py).
AMRL_OTC_FILE = Path(
    os.getenv(
        "AMRL_OTC_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_i.json"),
    )
)
# Anlage III: die Verordnungseinschränkungen und -ausschlüsse — das Gegenstück
# zu Anlage I. Lookup wie die übrigen, aber die Negativauskunft trägt hier
# *nicht*: die Anlage ist eine Übersicht und nennt sich nirgends abschließend
# (siehe src/amrl_verordnungsausschluss.py).
AMRL_VERORDNUNG_FILE = Path(
    os.getenv(
        "AMRL_VERORDNUNG_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_iii.json"),
    )
)
# Anlage II: die Lifestyle-Arzneimittel. Von den drei Listen zur
# Verordnungsfähigkeit ist ihr Ausschluss der härteste — er ruht auf § 34
# Absatz 1 Satz 7 SGB V, und der medizinisch begründete Einzelfall des § 16
# Absatz 5 AM-RL führt an ihm vorbei. Die Negativauskunft trägt hier wie bei
# Anlage III nicht (siehe src/amrl_lifestyle.py).
AMRL_LIFESTYLE_FILE = Path(
    os.getenv(
        "AMRL_LIFESTYLE_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_ii.json"),
    )
)
# Anlage IIa: die einzige Ausnahme von Anlage II. Sie gehört mit ihr zusammen —
# der Ausschluss der Raucherentwöhnung und der Anspruch auf Tabakentwöhnung nach
# § 34 Absatz 2 SGB V sind dieselbe Frage von zwei Seiten. Anders als bei Anlage
# II trägt die Negativauskunft hier: § 14a Absatz 3 Satz 1 AM-RL zählt die
# Ausnahme abschließend auf (siehe src/amrl_tabakentwoehnung.py).
AMRL_TABAKENTWOEHNUNG_FILE = Path(
    os.getenv(
        "AMRL_TABAKENTWOEHNUNG_FILE",
        str(BASE_DIR / "data" / "amrl_anlage_iia.json"),
    )
)

_PDF_ENV_DIR = os.getenv("PDF_DIR")

PDF_SEARCH_DIRS: List[Path] = [
    path
    for path in [
        Path(_PDF_ENV_DIR).resolve() if _PDF_ENV_DIR else None,
        DATA_DIR / "pdfs",
        BASE_DIR / "data" / "pdfs",
        BASE_DIR / "data",
    ]
    if path is not None
]

_PDF_NAME_RE = re.compile(r"^[\w.\- ]+\.pdf$", re.I)


SESSION_COOKIE = os.getenv("SESSION_COOKIE", "session_id")
SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "false").lower() == "true"
SESSION_COOKIE_MAX_AGE = int(os.getenv("SESSION_COOKIE_MAX_AGE", str(60 * 60 * 8)))


# Keine lokalen Windows-Pfade im Deployment.
# Standard: /data/chroma_db für Spaces; per Env überschreibbar.
# Wichtig: Die Chroma-DB liegt im Unterordner chroma_db (dorthin lädt der
# HF-Snapshot, siehe CHROMA_DATA_DIR). DATA_DIR selbst enthält nur eine leere
# chroma.sqlite3 -> "verfügbare Collections=[]".
CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", str(CHROMA_DATA_DIR))

if not Path(CHROMA_PERSIST_DIR).is_absolute():
    CHROMA_PERSIST_DIR = str((BASE_DIR / CHROMA_PERSIST_DIR).resolve())

CHROMA_COLLECTION = os.getenv("CHROMA_COLLECTION", "rv129")


# --- DEBUG: effektive Pfade prüfen, nur temporär ---

try:
    print("DEBUG BASE_DIR:", BASE_DIR)
    print("DEBUG STATIC_DIR:", STATIC_DIR)
    print("DEBUG INDEX_FILE:", INDEX_FILE)
    print("DEBUG effective CHROMA_PERSIST_DIR:", CHROMA_PERSIST_DIR)

    if Path("/data").exists():
        print("DEBUG /data listing:", os.listdir("/data")[:50])
        print("DEBUG exists /data/chroma.sqlite3:", (Path("/data") / "chroma.sqlite3").exists())

    print(
        "DEBUG exists CHROMA_PERSIST_DIR/chroma.sqlite3:",
        (Path(CHROMA_PERSIST_DIR) / "chroma.sqlite3").exists(),
    )
except Exception as e:
    print("DEBUG path check failed:", repr(e))

# --- /DEBUG ---


DEFAULT_CONTAINER_ID = os.getenv("DEFAULT_CONTAINER_ID", "Vertrag")

# "auto" liest das Embedding-Modell aus der Chroma-Collection-Metadata, die die
# Ingest-Pipeline schreibt.  Damit kann die Query-Seite nie mit einem anderen
# Modell arbeiten als der Index (die Ursache stiller Ranking-Ausfälle bzw.
# harter Dimensionsfehler nach einem Modellwechsel wie MiniLM -> e5-base).
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "auto")

# Cross-Encoder-Reranking der fusionierten Kandidaten.  Hob das Hybrid-Retrieval
# im Golden-Eval (configs/eval_cases.json im Ingest-Repo) von 17/18 auf 18/18.
ENABLE_RERANKER = os.getenv("ENABLE_RERANKER", "true").lower() == "true"
RERANKER_MODEL = os.getenv("RERANKER_MODEL", "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1")
RERANKER_CANDIDATES = int(os.getenv("RERANKER_CANDIDATES", "20"))

GROQ_MODEL = os.getenv("GROQ_MODEL", "openai/gpt-oss-120b")
GROQ_TEMPERATURE = float(os.getenv("GROQ_TEMPERATURE", "0.05"))
# Bei einem Reasoning-Modell (openai/gpt-oss-*) teilen sich Denk- und
# Antworttokens dieses Budget. Mit 1400 lief jede Antwort in `finish_reason:
# length`, und wenn das Denken es aufbrauchte, kam gar keine — gemessen am
# 19.08.2026: 1046 von 1400 Tokens allein fürs Reasoning. Der Client meldet die
# Kürzung jetzt im Log und behandelt eine leere Antwort als Fehler.
#
# Nach oben begrenzt das **Groq-Kontingent**, nicht das Modell: die
# Organisation liegt auf dem on_demand-Tier bei 8.000 Tokens pro Minute, und
# geprüft wird je Anfrage über Prompt **plus** max_tokens. Ein voller Kontext
# (DEFAULT_RAG_CONTEXT_CHARS=12.000) sind gemessen rund 4.825 Prompt-Tokens;
# 4.000 Antworttokens ergaben 8.825 und damit HTTP 413. 2.800 lässt gut 350
# Tokens Luft. Wer mehr braucht, braucht einen höheren Tier — oder einen
# kleineren Kontext.
GROQ_MAX_TOKENS = int(os.getenv("GROQ_MAX_TOKENS", "2800"))
# Nachschlag bei leerer Antwort. Ohne höheres Kontingent ist er ausgeschaltet
# (Decke = Budget): ein zweiter Aufruf mit mehr Tokens liefe in denselben 413.
GROQ_MAX_TOKENS_CEILING = int(os.getenv("GROQ_MAX_TOKENS_CEILING", str(GROQ_MAX_TOKENS)))
GROQ_DEBUG_PROMPTS = os.getenv("GROQ_DEBUG_PROMPTS", "false").lower() == "true"

DEFAULT_TOP_K = int(os.getenv("DEFAULT_TOP_K", "6"))
DEFAULT_FETCH_K = int(os.getenv("DEFAULT_FETCH_K", "18"))
DEFAULT_MAX_FINAL_RESULTS = int(os.getenv("DEFAULT_MAX_FINAL_RESULTS", "10"))
DEFAULT_MAX_SOURCES = int(os.getenv("DEFAULT_MAX_SOURCES", "5"))
# Garantierte Trefferplätze je abgefragtem Korpus, bevor global gekürzt wird.
# Ohne diese Untergrenze verliert ein kleines Korpus strukturell: die 106 Chunks
# der Abrechnungsvereinbarung kamen gegen die 6.670 des SGB V nie an, auch wenn
# die Regelung nur bei ihnen stand. Auf 0 gesetzt gilt wieder der reine
# Score-Schnitt. Reserviert wird höchstens die Hälfte der Ergebnisliste.
MIN_HITS_PER_CORPUS = int(os.getenv("MIN_HITS_PER_CORPUS", "2"))
# Wie viele Normen ein Lauf nachladen darf, weil der gefundene Text auf sie
# verweist. Auf 0 gesetzt ist die Verweisauflösung aus, und der Retriever
# verhält sich wie vorher.
#
# **Die Vorgabe ist seit dem 21.08.2026 die 0, und zwar gemessen.** Die Idee war
# richtig — ein einziger § des SGB V zitiert ein Dutzend andere, und was der Text
# selbst in Bezug nimmt, ist ein Signal, das die Ähnlichkeitssuche nicht hat.
# Nur ist das Budget fest: ein nachgeladener Verweis *ergänzt* den Kontext
# nicht, er *ersetzt* etwas.
#
#   19.08.2026, VERWEIS_SCORE 0.97 — Kettenabdeckung 85 % -> 75 %.
#   21.08.2026, VERWEIS_SCORE 0.50 — auf der damaligen Suite unauffällig (beide
#     Konfigurationen 100 %), weil die Suite gesättigt war und nichts mehr
#     trennen konnte. Auf der um fünf Fälle erweiterten Suite: 96,2 % ohne
#     Verweise, 93,6 % mit.
#
# Sichtbar wird es an `kette_off_label_abgabe`: nachgeladen wurden § 3 SGB V
# (solidarische Finanzierung) und § 300 SGB V (Abrechnung), beide ohne Bezug zur
# Frage und beide mit dem festen Verweis-Score 0.50 — verdrängt hat das
# § 31 Abs. 1 SGB V mit Score 0.871. Kein Fall der Suite gewinnt umgekehrt durch
# die Verweise ein Kettenglied.
#
# Modul und Messschalter (`--ohne-verweise`) bleiben: die Frage ist beantwortet,
# nicht erledigt. Wer den Verweis wieder aufnimmt, braucht ein eigenes Budget
# **neben** dem Top-k, keinen Platz darin — und muss ihn nach Einschlägigkeit
# auswählen statt nach Häufigkeit im Text.
MAX_VERWEIS_NORMEN = int(os.getenv("MAX_VERWEIS_NORMEN", "0"))
DEFAULT_RAG_CONTEXT_CHARS = int(os.getenv("DEFAULT_RAG_CONTEXT_CHARS", "12000"))
DEFAULT_MIN_SCORE = float(os.getenv("DEFAULT_MIN_SCORE", "0.0"))


# Wenn True, werden nur Quellen zurückgegeben, die das Modell als [Quelle n]
# in der Antwort erwähnt. Wenn das Modell keine Marker verwendet, fällt die App
# auf die kuratierte Quellenliste zurück.
SOURCE_SYNC_TO_ANSWER_MARKERS = os.getenv("SOURCE_SYNC_TO_ANSWER_MARKERS", "true").lower() == "true"


# --- Ask-Graph-Orchestrierung ---
# Organisiert die bestehende /ask-Funktionalität als expliziten Graphen.
# Wichtig: Es werden keine neuen fachlichen Routen eingeführt.

ASK_GRAPH_ENABLED = os.getenv("ASK_GRAPH_ENABLED", "true").lower() == "true"
ORCHESTRATOR_ENABLED = os.getenv("ORCHESTRATOR_ENABLED", "true").lower() == "true"
MAX_ASK_GRAPH_STEPS = int(os.getenv("MAX_ASK_GRAPH_STEPS", "10"))


_raw_origins = os.getenv("CORS_ORIGINS", "*")

CORS_ORIGINS = (
    ["*"]
    if _raw_origins.strip() == "*"
    else [origin.strip() for origin in _raw_origins.split(",") if origin.strip()]
)


# -----------------------------------------------------------------------------
# FastAPI-App
# -----------------------------------------------------------------------------

app = FastAPI(
    title=APP_TITLE,
    description="Retriever + Groq LLM + juristische Dokumente",
    version=APP_VERSION,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Static-Dateien ausliefern:
#
# /static/index.html
# /static/styles/style.css
# /static/js/script.js
# /static/berliner_apotheker_verein_cover.jpg
if os.path.isdir(STATIC_DIR):
    app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")


# -----------------------------------------------------------------------------
# Session State
# -----------------------------------------------------------------------------

@dataclass
class SessionState:
    """
    Pro-Session-Zustand.

    ConversationMemory lebt bewusst in der Session, nicht im GroqClient.
    Der GroqClient wird pro Request neu instanziiert und erhält memory nur explizit.
    """

    system_prompt: str = DEFAULT_SYSTEM_PROMPT
    memory: Optional[ConversationMemory] = None

    def get_memory(self) -> ConversationMemory:
        if self.memory is None:
            self.memory = ConversationMemory()
        return self.memory

    def reset(self) -> None:
        self.system_prompt = DEFAULT_SYSTEM_PROMPT
        self.get_memory().reset()


sessions: Dict[str, SessionState] = {}
sessions_lock = Lock()


def get_or_create_session(session_id: Optional[str]) -> tuple[str, SessionState]:
    with sessions_lock:
        if not session_id or session_id not in sessions:
            session_id = str(uuid4())
            sessions[session_id] = SessionState()
        return session_id, sessions[session_id]


@app.middleware("http")
async def session_middleware(request: Request, call_next):
    session_id, session = get_or_create_session(request.cookies.get(SESSION_COOKIE))
    request.state.session_id = session_id
    request.state.session = session

    response: Response = await call_next(request)
    response.set_cookie(
        SESSION_COOKIE,
        session_id,
        httponly=True,
        samesite="lax",
        secure=SESSION_COOKIE_SECURE,
        max_age=SESSION_COOKIE_MAX_AGE,
    )
    return response


# -----------------------------------------------------------------------------
# RAG-Komponenten
# -----------------------------------------------------------------------------

corpus_registry: Optional[CorpusRegistry] = None


def get_corpus_registry() -> CorpusRegistry:
    global corpus_registry
    if corpus_registry is None:
        corpus_registry = build_registry(
            persist_dir=CHROMA_PERSIST_DIR,
            model_name=EMBEDDING_MODEL,
            enable_reranker=ENABLE_RERANKER,
            reranker_model=RERANKER_MODEL,
            reranker_candidates=RERANKER_CANDIDATES,
        )
    return corpus_registry


def _build_retriever() -> FederatedRetriever:
    """
    Baut den Retriever über alle konfigurierten Korpora und gibt Startdiagnose aus.

    Bei genau einem Korpus verhält sich der föderierende Retriever wie der
    bisherige `LegalRetriever` — die Routing- und Fusionsschritte sind dann
    Durchreichen.
    """
    registry = get_corpus_registry()
    instance = FederatedRetriever(
        registry,
        router=route_question,
        # Die maßgeblichen Normen werden abgerufen, nicht erhofft. Das Register
        # liefert nur Adressen; der Retriever bleibt frei von Rechtsinhalten.
        anchors=norm_anchors.pflichtabruf,
        # Und die Normen, die der gefundene Text selbst in Bezug nimmt. Das
        # Register kennt die Kette eines Fragetyps, der Text die Kette seines
        # eigenen Satzes — „nach Maßgabe der §§ 11 bis 14" löste bis dahin
        # nichts aus. Seit dem 21.08.2026 steht `MAX_VERWEIS_NORMEN` auf 0 und
        # damit ist der Abruf aus; die Begründung steht dort.
        verweise=norm_verweise.abrufziele,
        # Nennt die Frage ein Regelwerk, führt dessen Treffer die Liste an —
        # auch gegen den Pflichtabruf. Der Anker behält seinen Platz im
        # Kontext, aber nicht den ersten.
        named_corpora=mentioned_corpora,
        min_hits_per_corpus=MIN_HITS_PER_CORPUS,
        max_verweis_normen=MAX_VERWEIS_NORMEN,
    )

    print("APP USING CHROMA PATH:", CHROMA_PERSIST_DIR)
    for entry in registry.diagnostics()["corpora"]:
        status = entry.get("count") if entry.get("ok") else entry.get("error")
        print(f"APP CORPUS {entry['corpus_id']}: collection={entry['collection']} count={status}")

    warning = registry.assert_consistent_embedding()
    if warning:
        print("APP WARNING:", warning)

    return instance


retriever: Optional[FederatedRetriever] = None
retriever_lock = Lock()


def get_retriever() -> FederatedRetriever:
    """Initialisiert Chroma/SentenceTransformer erst beim ersten echten Zugriff.

    Vorteil im Deployment: FastAPI kann importieren und starten, auch wenn Chroma
    oder das Embedding-Modell beim Build/Cold-Start kurz zicken. Der konkrete
    Fehler kommt dann kontrolliert über /health, /debug/retriever oder /ask.
    """
    global retriever
    if retriever is not None:
        return retriever

    with retriever_lock:
        if retriever is None:
            retriever = _build_retriever()
        return retriever


# -----------------------------------------------------------------------------
# Data Models
# -----------------------------------------------------------------------------

class Question(BaseModel):
    question: str = Field(..., min_length=1)

    top_k: Optional[int] = Field(default=None, ge=1, le=50)
    fetch_k: Optional[int] = Field(default=None, ge=1, le=100)
    max_final_results: Optional[int] = Field(default=None, ge=1, le=80)
    max_sources: Optional[int] = Field(default=None, ge=1, le=20)

    include_neighbors: bool = Field(default=True)
    include_explicit_sections: bool = Field(default=True)
    # False: Anlagen und Anhänge (z. B. Anlage 11, Dienstleistungs-Anhänge)
    # werden mitdurchsucht.  Die Priorisierung des Hauptvertrags übernimmt das
    # Ranking/Reranking, nicht ein harter Container-Filter, der früher ganze
    # Fragenklassen (Beitritt, pharmazeutische Dienstleistungen) blind machte.
    restrict_to_default_container: bool = Field(default=False)
    verify_negative_answer: bool = Field(default=True)
    allow_clarification: bool = Field(default=True)
    enrich_citations: bool = Field(default=True)
    debug: bool = Field(default=False)


class SystemPromptPayload(BaseModel):
    system_prompt: str = Field(..., min_length=1)


# -----------------------------------------------------------------------------
# Hilfsfunktionen
# -----------------------------------------------------------------------------

def _clean_question(text: str) -> str:
    return " ".join((text or "").strip().split())


def _build_llm(session: SessionState) -> GroqClient:
    return GroqClient(
        model=GROQ_MODEL,
        system_prompt=session.system_prompt,
        temperature=GROQ_TEMPERATURE,
        max_tokens=GROQ_MAX_TOKENS,
        max_tokens_ceiling=GROQ_MAX_TOKENS_CEILING,
        debug_prompts=GROQ_DEBUG_PROMPTS,
    )


def _build_composer(llm: GroqClient, payload: Question) -> AnswerComposer:
    """Baut den AnswerComposer im finalen Legal-RAG-Modus.

    Wichtig: Neue Composer-Versionen liefern nummernstabile, gruppierte Quellen
    mit Feldern wie display_title, source_numbers und canonical_refs. Diese
    Optionen werden hier bewusst explizit gesetzt, damit die API-/UI-Schicht
    nicht wieder auf die alte path/pages-Anzeige zurückfällt.
    """
    allowed = [DEFAULT_CONTAINER_ID] if payload.restrict_to_default_container else None

    kwargs: Dict[str, Any] = {
        "max_context_chars": DEFAULT_RAG_CONTEXT_CHARS,
        "pass_memory_to_llm_for_documents": False,
        "allowed_container_ids": allowed,
        "max_hits_for_context": payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS,
        "max_chunks_per_section": 4,
        "max_sources": payload.max_sources or DEFAULT_MAX_SOURCES,
        "min_score_for_context": DEFAULT_MIN_SCORE,
        "include_neighbor_hits_in_context": True,
        "include_pure_neighbors_as_sources": False,
        "append_sources_to_answer": False,
        # Neue Qualitätsoptionen des finalen Composer.
        "display_all_context_sources": True,
        "validate_source_markers": True,
        "prefer_direct_hits_over_neighbors": True,
    }

    try:
        return AnswerComposer(llm, **kwargs)
    except TypeError as exc:
        logger.warning(
            "AnswerComposer unterstützt nicht alle finalen Optionen; "
            "falle auf kompatible Minimalinitialisierung zurück: %s",
            exc,
        )
        legacy_keys = {
            "max_context_chars",
            "pass_memory_to_llm_for_documents",
            "allowed_container_ids",
            "max_hits_for_context",
            "max_chunks_per_section",
            "max_sources",
            "min_score_for_context",
            "include_neighbor_hits_in_context",
            "include_pure_neighbors_as_sources",
            "append_sources_to_answer",
        }
        legacy_kwargs = {key: value for key, value in kwargs.items() if key in legacy_keys}
        try:
            return AnswerComposer(llm, **legacy_kwargs)
        except TypeError as exc2:
            logger.warning(
                "AnswerComposer unterstützt nur Minimalparameter; "
                "Quellenqualität kann eingeschränkt sein: %s",
                exc2,
            )
            return AnswerComposer(
                llm,
                max_context_chars=DEFAULT_RAG_CONTEXT_CHARS,
                pass_memory_to_llm_for_documents=False,
            )


def _build_orchestrator(
    retriever_instance: LegalRetriever,
    composer: AnswerComposer,
    payload: Question,
) -> LegalAnswerOrchestrator:
    options = OrchestratorOptions(
        top_k=payload.top_k or DEFAULT_TOP_K,
        fetch_k=payload.fetch_k or DEFAULT_FETCH_K,
        max_final_results=payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS,
        max_sources=payload.max_sources or DEFAULT_MAX_SOURCES,
        include_neighbors=payload.include_neighbors,
        include_explicit_sections=payload.include_explicit_sections,
        restrict_to_default_container=payload.restrict_to_default_container,
        default_container_id=DEFAULT_CONTAINER_ID,
        min_score=DEFAULT_MIN_SCORE,
        enable_negative_recheck=payload.verify_negative_answer,
        enable_clarification=payload.allow_clarification,
        enrich_citations_with_canonical_refs=payload.enrich_citations,
        debug=payload.debug,
    )
    try:
        return LegalAnswerOrchestrator(
            retriever_instance,
            composer,
            options=options,
            question_classifier=classify_question,
            meta_detector=is_meta_question,
        )
    except TypeError:
        return LegalAnswerOrchestrator(retriever_instance, composer, options=options)


def _retrieve_hits(payload: Question) -> List[Dict[str, Any]]:
    question = _clean_question(payload.question)
    top_k = payload.top_k or DEFAULT_TOP_K
    fetch_k = payload.fetch_k or DEFAULT_FETCH_K
    max_final_results = payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS

    where = {"container_id": DEFAULT_CONTAINER_ID} if payload.restrict_to_default_container else None
    retriever_instance = get_retriever()

    try:
        return retriever_instance.query(
            question=question,
            top_k=top_k,
            where=where,
            fetch_k=fetch_k,
            include_explicit_sections=payload.include_explicit_sections,
            include_neighbors=payload.include_neighbors,
            max_final_results=max_final_results,
        )
    except TypeError:
        try:
            return retriever_instance.query(question=question, top_k=top_k, where=where)
        except TypeError:
            return retriever_instance.query(question=question, top_k=top_k)


def _coalesce(*values: Any) -> Any:
    for value in values:
        if value is not None and value != "":
            return value
    return None


def _page_value(hit: Dict[str, Any]) -> Optional[str]:
    """Normalisiert Seitenangaben aus Raw-Hits oder Composer-Sources."""
    metadata = hit.get("metadata") or {}

    direct = _coalesce(
        hit.get("pages"),
        hit.get("page_range"),
        metadata.get("pages"),
        metadata.get("page_range"),
    )
    if direct is not None:
        return str(direct)

    start = _coalesce(hit.get("page_start"), metadata.get("page_start"))
    end = _coalesce(hit.get("page_end"), metadata.get("page_end"), start)

    if start is not None and end is not None:
        return f"{start}–{end}"
    if start is not None:
        return str(start)
    return None


def _list_value(*values: Any) -> List[Any]:
    """Gibt die erste nicht-leere Listen-/Skalarangabe als Liste zurück."""
    for value in values:
        if value is None or value == "":
            continue
        if isinstance(value, (list, tuple, set)):
            return [item for item in value if item is not None and item != ""]
        return [value]
    return []


def _int_list(*values: Any) -> List[int]:
    out: List[int] = []
    for value in _list_value(*values):
        try:
            number = int(value)
        except (TypeError, ValueError):
            continue
        if number not in out:
            out.append(number)
    return sorted(out)


def _int_or_none(value: Any) -> Optional[int]:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _highlight_text_from_hit(hit: Dict[str, Any], *, max_chars: int = 2000) -> str:
    """Chunk-Text für die PDF-Hervorhebung, ohne konstruierten Kontext-Header.

    Die Ingest-Pipeline stellt jedem Chunk eine Zeile "Container · § x Titel"
    voran, die im PDF nicht existiert und daher nicht gematcht werden kann.
    """
    text = str(hit.get("text") or hit.get("document") or "").strip()
    if not text:
        return ""
    head, sep, rest = text.partition("\n\n")
    if sep and " · " in head and len(head) <= 200 and rest.strip():
        text = rest.strip()
    return text[:max_chars]


def _highlights_from_hit(hit: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Normalisiert die Hervorhebungsliste einer Quelle bzw. eines Raw-Hits."""
    metadata = hit.get("metadata") or {}

    existing = hit.get("highlights")
    if isinstance(existing, list) and existing:
        out = []
        for entry in existing:
            if not isinstance(entry, dict):
                continue
            text = str(entry.get("text") or "").strip()
            if not text:
                continue
            out.append(
                {
                    "page_start": _int_or_none(entry.get("page_start")),
                    "page_end": _int_or_none(entry.get("page_end")) or _int_or_none(entry.get("page_start")),
                    "text": text[:2000],
                }
            )
        return out

    text = _highlight_text_from_hit(hit)
    if not text:
        return []

    page_start = _int_or_none(_coalesce(hit.get("page_start"), metadata.get("page_start")))
    page_end = _int_or_none(_coalesce(hit.get("page_end"), metadata.get("page_end"))) or page_start
    return [{"page_start": page_start, "page_end": page_end, "text": text}]


def _canonical_refs_from_hit(hit: Dict[str, Any]) -> List[str]:
    metadata = hit.get("metadata") or {}
    refs = [
        str(ref).strip()
        for ref in _list_value(hit.get("canonical_refs"), metadata.get("canonical_refs"))
        if str(ref).strip()
    ]

    single = _coalesce(hit.get("canonical_ref"), metadata.get("canonical_ref"))
    if single is not None and str(single).strip() and str(single).strip() not in refs:
        refs.append(str(single).strip())

    return refs


def _source_marker(numbers: List[int]) -> str:
    if not numbers:
        return ""
    if len(numbers) == 1:
        return f"[Quelle {numbers[0]}]"
    return "[Quellen " + ", ".join(str(n) for n in numbers) + "]"


_SGB_BOOK_RE = re.compile(r"Sozialgesetzbuch.*?\(([IVXLC]+)\)", re.I)


def _short_doc_title(title: str, *, max_chars: int = 32) -> str:
    """Kürzt einen Dokumenttitel auf eine zitierfähige Kurzform.

    Die vollen Titel aus den Chunk-Metadaten sind für eine Quellenzeile zu lang
    ("Sozialgesetzbuch (SGB) Fünftes Buch (V) - Gesetzliche Krankenversicherung").
    Das LLM bekommt weiterhin den vollen Titel im Kontext; gekürzt wird nur die
    Anzeige.
    """
    text = " ".join(str(title or "").split())
    if not text:
        return ""

    match = _SGB_BOOK_RE.search(text)
    if match:
        return f"SGB {match.group(1).upper()}"

    # Untertitel nach Gedankenstrich abschneiden.
    text = re.split(r"\s+[–—-]\s+", text, maxsplit=1)[0].strip()
    if text.lower().startswith("rahmenvertrag"):
        return "Rahmenvertrag"
    if len(text) <= max_chars:
        return text
    return text[:max_chars].rsplit(" ", 1)[0].rstrip(" ,;:-–—") + "…"


def _format_source_display_title(source: Dict[str, Any]) -> str:
    """Erzeugt eine UI-fertige, nummernstabile Quellenanzeige."""
    marker = _coalesce(source.get("source_label"), source.get("source_marker")) or _source_marker(
        _int_list(source.get("source_numbers"), source.get("source_number"))
    )

    container = _coalesce(source.get("container"), "Unbekannt")
    section = _coalesce(source.get("section"), "ohne Abschnitt")
    path = _coalesce(source.get("path"), source.get("section_path"))
    if not path:
        path = f"{container}::{section}"

    # Dokumentangabe voranstellen, sobald der Korpus mehr als ein Dokument
    # enthält. "§ 16" allein ist mehrdeutig: Rahmenvertrag und SGB V teilen sich
    # 34 §-Nummern mit unterschiedlichem Inhalt.
    doc = str(_coalesce(source.get("doc_title"), source.get("doc_id")) or "").strip()
    doc_part = f"{_short_doc_title(doc)} · " if doc else ""

    pages = _coalesce(source.get("page_range"), source.get("pages"), "?")
    canonical_refs = [
        str(ref).strip()
        for ref in _list_value(source.get("canonical_refs"), source.get("canonical_ref"))
        if str(ref).strip()
    ]

    # Keine redundante Anzeige "§ 6 (§ 6)".
    canonical_refs = [ref for ref in dict.fromkeys(canonical_refs) if ref != section]
    ref_part = f" ({'; '.join(canonical_refs[:4])})" if canonical_refs else ""

    role = ""
    kinds = set(source.get("retrieval_kinds") or [])
    if kinds == {"neighbor"}:
        role = " · Kontext/Nachbar"

    return f"{marker} {doc_part}{path}{ref_part}, Seiten {pages}{role}".strip()


def _source_from_hit(hit: Dict[str, Any], *, source_number: Optional[int] = None) -> Dict[str, Any]:
    """
    Einheitliche Quellennormalisierung.

    Unterstützt Raw-Retriever-Hits und bereits kuratierte Composer-Quellen.
    Anders als die ältere Version bewahrt sie Composer-Felder wie
    display_title, source_numbers und canonical_refs, damit die UI nicht wieder
    auf bloße path/pages-Ausgaben zurückfällt.
    """
    metadata = hit.get("metadata") or {}

    explicit_number = source_number if source_number is not None else _coalesce(
        hit.get("source_number"),
        metadata.get("source_number"),
    )
    source_numbers = _int_list(
        hit.get("source_numbers"),
        metadata.get("source_numbers"),
        explicit_number,
    )

    pages = _page_value(hit)
    container = _coalesce(hit.get("container"), hit.get("container_id"), metadata.get("container_id"))
    section = _coalesce(hit.get("section"), hit.get("section_id"), metadata.get("section_id"))
    canonical_refs = _canonical_refs_from_hit(hit)
    canonical_ref = _coalesce(hit.get("canonical_ref"), metadata.get("canonical_ref"))
    if canonical_ref is None and canonical_refs:
        canonical_ref = canonical_refs[0]

    path = _coalesce(
        hit.get("path"),
        hit.get("section_path"),
        metadata.get("path"),
        metadata.get("section_path"),
    )
    if not path and container and section:
        path = f"{container}::{section}"

    source: Dict[str, Any] = {
        "source_number": explicit_number,
        "source_numbers": source_numbers,
        "source_marker": _coalesce(hit.get("source_marker"), metadata.get("source_marker"), _source_marker(source_numbers)),
        "source_label": _coalesce(hit.get("source_label"), metadata.get("source_label"), _source_marker(source_numbers)),
        "container": container,
        "section": section,
        "canonical_ref": canonical_ref,
        "canonical_refs": canonical_refs,
        "path": path,
        "pages": pages,
        "page_range": pages,
        "chunk_index": _coalesce(
            hit.get("chunk_index"),
            hit.get("chunk_index_in_section"),
            metadata.get("chunk_index"),
            metadata.get("chunk_index_in_section"),
        ),
        # PDF-Locator für die interaktive Fundstellen-Anzeige in der UI.
        "page_start": _int_or_none(_coalesce(hit.get("page_start"), metadata.get("page_start"))),
        "page_end": _int_or_none(_coalesce(hit.get("page_end"), metadata.get("page_end"), hit.get("page_start"), metadata.get("page_start"))),
        "source_file": _coalesce(hit.get("source_file"), metadata.get("source_file")),
        "doc_id": _coalesce(hit.get("doc_id"), metadata.get("doc_id")),
        "doc_title": _coalesce(hit.get("doc_title"), metadata.get("doc_title")),
        # Von FederatedRetriever je Treffer gesetzt. Ohne Durchreichen weiß die
        # API-Schicht nicht mehr, aus welchem Korpus eine Quelle stammt — der
        # Änderungshinweis in corpus_amendments.py könnte dann nie greifen.
        "corpus_id": _coalesce(hit.get("corpus_id"), metadata.get("corpus_id")),
        "highlights": _highlights_from_hit(hit),
        "legal_unit_id": _coalesce(hit.get("legal_unit_id"), metadata.get("legal_unit_id")),
        "parent_unit_id": _coalesce(hit.get("parent_unit_id"), metadata.get("parent_unit_id")),
        "chunk_kind": _coalesce(hit.get("chunk_kind"), metadata.get("chunk_kind")),
        "unit_type": _coalesce(hit.get("unit_type"), metadata.get("unit_type")),
        "score": hit.get("score"),
        "rank_score": hit.get("rank_score"),
        "retrieval_kinds": hit.get("retrieval_kinds", metadata.get("retrieval_kinds", [])),
        "display_label": _coalesce(hit.get("display_label"), metadata.get("display_label")),
        "display_title": _coalesce(hit.get("display_title"), metadata.get("display_title")),
    }

    if not source["display_title"]:
        source["display_title"] = _format_source_display_title(source)
    if not source["display_label"]:
        source["display_label"] = source["display_title"]

    return source


def _extract_answer_source_numbers(answer: str) -> List[int]:
    """
    Liest Quellenmarker aus der Modellantwort:
        [Quelle 1], [Quelle 2], ...
    """
    nums: List[int] = []
    for m in re.finditer(r"\[Quelle\s+(\d+)(?:[^\]]*)\]", answer or "", flags=re.I):
        try:
            nums.append(int(m.group(1)))
        except ValueError:
            continue
    return list(dict.fromkeys(nums))


def _answer_reports_nothing_found(answer: str) -> bool:
    """Die Antwort zitiert nichts, nennt keine Norm und sagt, es gebe nichts.

    Zwei Stellen hängen an dieser Frage: der Reichweiten-Hinweis, dessen Prämisse
    ist, dass Textstellen herangezogen *wurden*, und die zurückgegebene
    Quellenliste, die sonst fünf Fundstellen unter eine Aussage setzt, dass es
    keine gibt.

    Bewusst eng gefasst, damit der gewollte Fallback erhalten bleibt: vergisst das
    Modell bei einer inhaltlichen Antwort die Marker, sollen seine Quellen weiter
    erscheinen. Ein § irgendwo im Text heißt deshalb "etwas gefunden", unabhängig
    davon, was die Antwort darüber behauptet.
    """
    text = answer or ""
    if _extract_answer_source_numbers(text):
        return False
    if "§" in text:
        return False
    return bool(NEGATIVE_ANSWER_RE.search(text))


def _merge_source(existing: Dict[str, Any], incoming: Dict[str, Any]) -> Dict[str, Any]:
    """Führt gruppierbare Quellen zusammen, ohne Nummern/Fundstellen zu verlieren."""
    merged = dict(existing)

    numbers = _int_list(existing.get("source_numbers"), existing.get("source_number"), incoming.get("source_numbers"), incoming.get("source_number"))
    refs = [
        str(ref).strip()
        for ref in _list_value(existing.get("canonical_refs"), existing.get("canonical_ref"), incoming.get("canonical_refs"), incoming.get("canonical_ref"))
        if str(ref).strip()
    ]
    refs = list(dict.fromkeys(refs))

    kinds = list(dict.fromkeys(
        list(existing.get("retrieval_kinds") or []) + list(incoming.get("retrieval_kinds") or [])
    ))

    merged["source_numbers"] = numbers
    merged["source_number"] = numbers[0] if numbers else _coalesce(existing.get("source_number"), incoming.get("source_number"))
    merged["source_marker"] = _source_marker(numbers) or _coalesce(existing.get("source_marker"), incoming.get("source_marker"))
    merged["source_label"] = merged["source_marker"]
    merged["canonical_refs"] = refs
    merged["canonical_ref"] = _coalesce(existing.get("canonical_ref"), incoming.get("canonical_ref"), refs[0] if refs else None)
    merged["retrieval_kinds"] = kinds

    # Behalte den höchsten Score, ohne None/Strings zu erzwingen.
    for key in ("score", "rank_score"):
        try:
            old_score = float(existing.get(key) or 0.0)
            new_score = float(incoming.get(key) or 0.0)
            merged[key] = max(old_score, new_score)
        except (TypeError, ValueError):
            merged[key] = _coalesce(existing.get(key), incoming.get(key))

    # Bei mehreren Chunks derselben Section ist chunk_index für die Anzeige nicht
    # mehr sinnvoll eindeutig. Nur behalten, wenn identisch.
    if existing.get("chunk_index") != incoming.get("chunk_index"):
        merged["chunk_index"] = None

    # PDF-Locator zusammenführen: Seitenbereich aufspannen, Datei behalten,
    # Hervorhebungen beider Quellen sammeln.
    starts = [p for p in (_int_or_none(existing.get("page_start")), _int_or_none(incoming.get("page_start"))) if p is not None]
    ends = [p for p in (_int_or_none(existing.get("page_end")), _int_or_none(incoming.get("page_end"))) if p is not None]
    merged["page_start"] = min(starts) if starts else None
    merged["page_end"] = max(ends) if ends else merged["page_start"]
    merged["source_file"] = _coalesce(existing.get("source_file"), incoming.get("source_file"))
    merged["doc_id"] = _coalesce(existing.get("doc_id"), incoming.get("doc_id"))
    merged["doc_title"] = _coalesce(existing.get("doc_title"), incoming.get("doc_title"))

    highlights: List[Dict[str, Any]] = []
    seen_highlights: set = set()
    for entry in list(existing.get("highlights") or []) + list(incoming.get("highlights") or []):
        if not isinstance(entry, dict):
            continue
        text = str(entry.get("text") or "").strip()
        if not text:
            continue
        dedupe_key = (entry.get("page_start"), text[:120])
        if dedupe_key in seen_highlights:
            continue
        seen_highlights.add(dedupe_key)
        highlights.append(entry)
    merged["highlights"] = highlights[:8]

    merged["display_title"] = _format_source_display_title(merged)
    merged["display_label"] = merged["display_title"]
    return merged


def _source_dedupe_key(src: Dict[str, Any]) -> tuple[Any, ...]:
    """Dedupe-Key, der Composer-Gruppierungen respektiert."""
    numbers = tuple(_int_list(src.get("source_numbers"), src.get("source_number")))
    if numbers:
        return ("numbers", numbers)

    display_title = src.get("display_title")
    if display_title:
        return ("display", display_title)

    canonical_refs = tuple(src.get("canonical_refs") or [])
    return (
        "location",
        # doc_id zuerst: gleiche §-Nummern in verschiedenen Dokumenten dürfen
        # nicht zu einer Quelle verschmelzen.
        src.get("doc_id"),
        src.get("container"),
        src.get("section"),
        src.get("pages") or src.get("page_range"),
        canonical_refs,
    )


def _dedupe_sources(sources: Iterable[Dict[str, Any]], *, max_sources: int) -> List[Dict[str, Any]]:
    """Dedupliziert Quellen, ohne Composer-Felder zu verlieren.

    Die alte Variante normalisierte jede Quelle auf path/pages/chunk_index zurück
    und zerstörte dadurch display_title, source_numbers und canonical_refs.
    """
    grouped: Dict[tuple[Any, ...], Dict[str, Any]] = {}
    order: List[tuple[Any, ...]] = []

    for raw in sources:
        src = _source_from_hit(raw)

        # Wenn keine Nummern vorhanden sind, gruppiere gleiche Norm/Seiten, nicht
        # einzelne Chunk-Indizes. Das verhindert doppelte "Vertrag::§ 6"-Zeilen.
        key = _source_dedupe_key(src)
        if key not in grouped:
            grouped[key] = src
            order.append(key)
        else:
            grouped[key] = _merge_source(grouped[key], src)

    out = [grouped[key] for key in order]
    out.sort(key=lambda s: (_int_list(s.get("source_numbers"), s.get("source_number")) or [10_000])[0])
    return out[:max_sources]


def _build_sources_from_composer_or_hits(
    composer: AnswerComposer,
    hits: List[Dict[str, Any]],
    payload: Question,
) -> List[Dict[str, Any]]:
    max_sources = payload.max_sources or DEFAULT_MAX_SOURCES
    allowed = [DEFAULT_CONTAINER_ID] if payload.restrict_to_default_container else None

    if hasattr(composer, "build_sources"):
        try:
            raw_sources = composer.build_sources(
                hits,
                max_sources=max(max_sources, min(len(hits), DEFAULT_MAX_FINAL_RESULTS)),
                include_pure_neighbors=False,
                allowed_container_ids=allowed,
            )
            return _dedupe_sources(raw_sources, max_sources=max_sources)
        except TypeError:
            try:
                raw_sources = composer.build_sources(hits, max_sources=max_sources)
                return _dedupe_sources(raw_sources, max_sources=max_sources)
            except Exception as exc:
                logger.warning("Composer build_sources fallback failed: %s", exc)

    return _dedupe_sources(hits, max_sources=max_sources)


def _sources_cover_cited_numbers(sources: List[Dict[str, Any]], cited_numbers: List[int]) -> bool:
    if not cited_numbers:
        return True
    covered = set()
    for source in sources:
        covered.update(_int_list(source.get("source_numbers"), source.get("source_number")))
    return set(cited_numbers).issubset(covered)


def _sources_for_cited_numbers_from_raw_hits(
    cited_numbers: List[int],
    hits: List[Dict[str, Any]],
    *,
    max_sources: int,
) -> List[Dict[str, Any]]:
    """Letzter Fallback: mappt [Quelle n] auf Treffer n.

    Dieser Pfad ist nur ein Sicherheitsnetz. Bevorzugt werden Composer-Sources,
    weil der Composer den Kontext sortiert/dedupliziert und die Nummern korrekt
    kennt.
    """
    cited_sources: List[Dict[str, Any]] = []
    for num in cited_numbers:
        idx = num - 1
        if 0 <= idx < len(hits):
            cited_sources.append(_source_from_hit(hits[idx], source_number=num))
    return _dedupe_sources(cited_sources, max_sources=max_sources)


_FINE_REF_RE = re.compile(
    r"§\s*(\d{1,3}[a-z]?)\s+Abs\.\s*(\d+[a-z]?)\s+Buchst\.\s*([a-z])",
    flags=re.I,
)


def _evidence_blob(sources: List[Dict[str, Any]], hits: List[Dict[str, Any]]) -> str:
    parts: List[str] = []
    for item in list(sources or []) + list(hits or []):
        metadata = item.get("metadata") or {}
        parts.extend(
            str(value)
            for value in [
                item.get("canonical_ref"),
                item.get("canonical_refs"),
                item.get("section"),
                item.get("section_id"),
                item.get("path"),
                item.get("text"),
                item.get("document"),
                metadata.get("canonical_ref"),
                metadata.get("section_id"),
                metadata.get("text"),
            ]
            if value
        )
    return " ".join(parts).lower()


def _sanitize_unsupported_fine_references(
    answer: str,
    *,
    sources: List[Dict[str, Any]],
    hits: List[Dict[str, Any]],
) -> str:
    """Entschärft erfundene Feinfundstellen wie '§ 6 Abs. 1 Buchst. a'.

    Wenn die genaue Buchstabenfundstelle nicht im Kontext/Metadaten belegt ist,
    wird auf die belastbarere Absatzfundstelle zurückgeführt.
    """
    if not answer:
        return answer

    evidence = _evidence_blob(sources, hits)

    def repl(match: re.Match[str]) -> str:
        para, abs_no, letter = match.group(1), match.group(2), match.group(3).lower()
        exact_patterns = [
            f"§ {para} abs. {abs_no} buchst. {letter}",
            f"§{para} abs. {abs_no} buchst. {letter}",
            f"§ {para} absatz {abs_no} buchstabe {letter}",
            f"§ {para} abs. {abs_no} lit. {letter}",
        ]
        if any(pattern in evidence for pattern in exact_patterns):
            return match.group(0)
        return f"§ {para} Abs. {abs_no}"

    cleaned = _FINE_REF_RE.sub(repl, answer)

    # Entfernt häufige Dopplung: "(§ 6 Abs. 1), § 6 Abs. 1"
    cleaned = re.sub(
        r"\((§\s*\d{1,3}[a-z]?\s+Abs\.\s*\d+[a-z]?)\),\s*\1",
        r"(\1)",
        cleaned,
        flags=re.I,
    )
    return cleaned


def _postprocess_answer(
    composer: AnswerComposer,
    answer: str,
    *,
    sources: List[Dict[str, Any]],
    hits: List[Dict[str, Any]],
) -> str:
    """Zentraler letzter Antwort-Postprocessor für Composer- und Orchestratorpfad."""
    text = answer or ""

    if hasattr(composer, "_strip_model_generated_sources"):
        try:
            text = composer._strip_model_generated_sources(text)  # type: ignore[attr-defined]
        except Exception as exc:
            logger.debug("Composer source-strip postprocessing skipped: %s", exc)

    if hasattr(composer, "_strip_invalid_source_markers"):
        try:
            # Erwartet meist used_hits/source_number-Strukturen; sources sind dafür
            # besser geeignet als Raw-Hits.
            text = composer._strip_invalid_source_markers(text, sources or hits)  # type: ignore[attr-defined]
        except Exception as exc:
            logger.debug("Composer marker postprocessing skipped: %s", exc)

    text = _sanitize_unsupported_fine_references(text, sources=sources, hits=hits)

    # Zuletzt, damit die vorherigen Schritte erst entfernen können, was einen
    # Abschnitt nur scheinbar füllt (ungültige Marker, unbelegte Fundstellen).
    if hasattr(composer, "_strip_empty_section_headings"):
        try:
            text = composer._strip_empty_section_headings(text)  # type: ignore[attr-defined]
        except Exception as exc:
            logger.debug("Composer section postprocessing skipped: %s", exc)

    return text.strip()


def _normalize_returned_sources(
    *,
    answer: str,
    sources: List[Dict[str, Any]],
    hits: List[Dict[str, Any]],
    payload: Question,
) -> List[Dict[str, Any]]:
    """
    Finale API-Quellenlogik.

    Wichtigste Regel:
    Composer-/Orchestrator-Sources sind der primäre Wahrheitsanker für [Quelle n].
    Raw-Hits werden nur als Fallback verwendet. Dadurch bleiben display_title,
    source_numbers und canonical_refs erhalten.
    """
    max_sources = payload.max_sources or DEFAULT_MAX_SOURCES
    cited_numbers = _extract_answer_source_numbers(answer)

    # Hat die Antwort nichts gefunden, gibt es auch nichts anzuzeigen. Die
    # kuratierte Liste wäre hier kein Fallback, sondern eine Behauptung.
    if _answer_reports_nothing_found(answer):
        return []

    normalized = _dedupe_sources(
        sources or [],
        max_sources=max(max_sources, len(cited_numbers), DEFAULT_MAX_SOURCES),
    )

    if normalized:
        if SOURCE_SYNC_TO_ANSWER_MARKERS and cited_numbers:
            # Zeige mindestens alle zitierten Quellen. Wenn Composer-Sources die
            # zitierten Nummern abdecken, keine Rückabbildung auf Raw-Hits.
            if _sources_cover_cited_numbers(normalized, cited_numbers):
                cited_set = set(cited_numbers)
                cited_first = [
                    src for src in normalized
                    if cited_set.intersection(_int_list(src.get("source_numbers"), src.get("source_number")))
                ]
                rest = [src for src in normalized if src not in cited_first]
                return (cited_first + rest)[:max(max_sources, len(cited_first))]
            fallback = _sources_for_cited_numbers_from_raw_hits(cited_numbers, hits, max_sources=max_sources)
            if fallback:
                return fallback
        return normalized[:max_sources]

    if SOURCE_SYNC_TO_ANSWER_MARKERS and cited_numbers and hits:
        fallback = _sources_for_cited_numbers_from_raw_hits(cited_numbers, hits, max_sources=max_sources)
        if fallback:
            return fallback

    return _dedupe_sources(hits, max_sources=max_sources)


_INLINE_MARKER_RE = re.compile(r"\[Quellen?\s+\d+(?:[^\]]*)\]", flags=re.I)


def _remap_inline_source_markers(answer: str, old_to_new: Dict[int, int]) -> str:
    """Schreibt Inline-Marker [Quelle n]/[Quellen n, m] auf die neuen Anzeigenummern um.

    Unbekannte Nummern (kein Mapping vorhanden) bleiben unverändert stehen; sie
    werden an anderer Stelle bereits als ungültige Marker entfernt.
    """
    if not answer or not old_to_new:
        return answer

    def repl(match: re.Match[str]) -> str:
        mapped: List[int] = []
        for raw in re.findall(r"\d+", match.group(0)):
            try:
                new = old_to_new.get(int(raw))
            except (TypeError, ValueError):
                new = None
            if new is not None and new not in mapped:
                mapped.append(new)
        if not mapped:
            return match.group(0)
        mapped.sort()
        return _source_marker(mapped)

    return _INLINE_MARKER_RE.sub(repl, answer)


def _renumber_sources_for_display(
    answer: str,
    sources: List[Dict[str, Any]],
) -> Tuple[str, List[Dict[str, Any]]]:
    """Vergibt stabile, fortlaufende Anzeigenummern (1..n) für die finalen Quellen.

    Hintergrund: Die ursprünglichen [Quelle n]-Nummern sind Retrieval-Ränge aus
    dem RAG-Kontext. Dadurch beginnt die Anzeige nicht bei 1 und enthält Lücken
    (z. B. "[Quellen 3, 8]"). Für die UI werden die tatsächlich angezeigten
    Quellen in ihrer Anzeige-Reihenfolge auf 1..n abgebildet und die
    Inline-Marker im Antworttext konsistent mitgezogen. Mehrere Alt-Nummern
    derselben Quelle (gruppierte Chunks) fallen dabei auf eine Anzeigenummer
    zusammen.
    """
    if not sources:
        return answer, sources

    old_to_new: Dict[int, int] = {}
    next_number = 1

    for source in sources:
        old_numbers = _int_list(source.get("source_numbers"), source.get("source_number"))
        if not old_numbers:
            # Quelle ohne Nummer: entsteht über den Rückfall auf die Roh-Treffer
            # (`_dedupe_sources(hits, …)`), denn ein Treffer trägt keine
            # Quellennummer — die vergibt erst der Composer. Übersprungen bekam
            # sie hier nie `display_title`/`display_label` und erschien in der
            # UI als **leere Zeile**. Sie bekommt deshalb ihre Anzeige, aber
            # keinen Marker: zitiert wurde sie nicht, und eine erfundene Nummer
            # stünde im Antworttext nirgends.
            source.setdefault("source_marker", "")
            source.setdefault("source_label", "")
            source["display_title"] = _format_source_display_title(source)
            source["display_label"] = source["display_title"]
            continue

        assigned: Optional[int] = None
        for old in old_numbers:
            if old in old_to_new:
                assigned = old_to_new[old]
                break
        if assigned is None:
            assigned = next_number
            next_number += 1
        for old in old_numbers:
            old_to_new.setdefault(old, assigned)

        marker = f"[Quelle {assigned}]"
        source["source_number"] = assigned
        source["source_numbers"] = [assigned]
        source["source_marker"] = marker
        source["source_label"] = marker
        source["display_title"] = _format_source_display_title(source)
        source["display_label"] = source["display_title"]

    new_answer = _remap_inline_source_markers(answer, old_to_new)
    return new_answer, sources


def _debug_hit(hit: Dict[str, Any]) -> Dict[str, Any]:
    text = (hit.get("text") or hit.get("document") or "").strip()
    return {
        **_source_from_hit(hit),
        "text_preview": text[:350],
        "text_length": len(text),
    }


def _collection_count() -> Any:
    """Chunks über alle Korpora. Bei einem Korpus identisch zum bisherigen Wert."""
    try:
        return get_corpus_registry().total_count()
    except Exception as exc:
        logger.warning("Chroma count unavailable: %s", exc)
        return "unknown"


def _corpus_counts() -> Dict[str, Any]:
    try:
        return {
            entry["corpus_id"]: entry.get("count", entry.get("error"))
            for entry in get_corpus_registry().diagnostics()["corpora"]
        }
    except Exception as exc:  # noqa: BLE001
        logger.warning("Korpus-Diagnose nicht verfügbar: %s", exc)
        return {}


# -----------------------------------------------------------------------------
# Ask Graph Workflow
# -----------------------------------------------------------------------------

AskRoute = Literal["meta", "non_meta", "legacy", "clarification"]
AskNodeName = Literal[
    "__start__",
    "meta_decision",
    "retrieve_hits",
    "compose_answer",
    "orchestrate_answer",
    "normalize_sources",
    "update_memory",
    "build_response",
    "__end__",
]


@dataclass
class AskGraphStep:
    """Rückgabe eines Graph-Knotens."""

    next_node: AskNodeName
    reason: str = ""


@dataclass
class AskGraphTraceEntry:
    node: str
    next_node: str
    reason: str = ""
    route: Optional[str] = None
    hit_count: int = 0
    answer_type: str = "unknown"


@dataclass
class AskGraphContext:
    """
    Gemeinsamer Zustand des /ask-Graphen.

    Dieser Graph kapselt ausschließlich die bereits vorhandene Funktionalität:
    Meta-Fragen werden ohne Retrieval beantwortet, alle anderen Fragen laufen
    durch Retrieval, AnswerComposer, Quellennormalisierung, Memory-Update und
    Response-Aufbau.
    """

    payload: Question
    request: Request
    session: SessionState
    memory: ConversationMemory
    llm: GroqClient
    composer: AnswerComposer
    question: str
    orchestrator: Optional[LegalAnswerOrchestrator] = None

    route: Optional[AskRoute] = None
    hits: Optional[List[Dict[str, Any]]] = None
    # Die Teilmenge von `hits`, die die Antwort auch zitiert. Wird im
    # Normalisierungsschritt gesetzt, solange der Antworttext noch die
    # Retrieval-Ränge trägt (siehe `AnswerComposer.cited_hits`).
    cited_hits: Optional[List[Dict[str, Any]]] = None
    raw_sources: Optional[List[Dict[str, Any]]] = None
    sources: Optional[List[Dict[str, Any]]] = None
    answer: str = ""
    answer_type: str = "unknown"
    response_body: Optional[Dict[str, Any]] = None
    orchestrator_debug: Optional[Dict[str, Any]] = None
    needs_clarification: bool = False
    clarification_question: Optional[str] = None
    trace: Optional[List[AskGraphTraceEntry]] = None

    def __post_init__(self) -> None:
        if self.hits is None:
            self.hits = []
        if self.raw_sources is None:
            self.raw_sources = []
        if self.sources is None:
            self.sources = []
        if self.cited_hits is None:
            self.cited_hits = []
        if self.trace is None:
            self.trace = []


def _ask_graph_meta_decision(ctx: AskGraphContext) -> AskGraphStep:
    """Routet Meta-Fragen, Orchestrator-Fragen und Legacy-Fallback sauber."""
    if is_meta_question(ctx.question):
        ctx.route = "meta"
        return AskGraphStep("compose_answer", "meta question without retrieval")

    if ORCHESTRATOR_ENABLED and ctx.orchestrator is not None:
        ctx.route = "non_meta"
        return AskGraphStep("orchestrate_answer", "regular question via legal orchestrator")

    ctx.route = "legacy"
    return AskGraphStep("retrieve_hits", "orchestrator unavailable or disabled; legacy retrieval")


def _ask_graph_retrieve_hits(ctx: AskGraphContext) -> AskGraphStep:
    """Entspricht dem bisherigen Aufruf von _retrieve_hits(payload)."""
    ctx.hits = _retrieve_hits(ctx.payload)
    return AskGraphStep("compose_answer", f"retrieved {len(ctx.hits)} hits")


def _ask_graph_orchestrate_answer(ctx: AskGraphContext) -> AskGraphStep:
    """Fachlicher Orchestrator-Knoten: Retrieval, Recheck, Antwort, Audit."""
    if ctx.orchestrator is None:
        ctx.route = "legacy"
        return AskGraphStep("retrieve_hits", "orchestrator unavailable; falling back to legacy retrieval")

    result = ctx.orchestrator.run(ctx.question, memory=ctx.memory)
    ctx.answer = result.answer
    ctx.answer_type = result.answer_type
    ctx.hits = result.hits
    ctx.raw_sources = result.raw_sources
    ctx.sources = result.sources
    ctx.needs_clarification = result.needs_clarification
    ctx.clarification_question = result.clarification_question
    ctx.orchestrator_debug = result.to_debug_dict()
    if result.needs_clarification:
        ctx.route = "clarification"
    return AskGraphStep("normalize_sources", "answer orchestrated and audited")


def _ask_graph_compose_answer(ctx: AskGraphContext) -> AskGraphStep:
    """
    Entspricht der bisherigen Antwortgenerierung im /ask-Endpoint.

    Meta-Fragen werden mit leerem Kontext komponiert. Nicht-Meta-Fragen nutzen
    weiterhin compose_with_sources, sofern vorhanden, sonst compose plus
    Quellen-Fallback.
    """
    if ctx.route == "meta":
        ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, [], memory=ctx.memory)
        ctx.raw_sources = []
        return AskGraphStep("normalize_sources", "meta answer composed")

    if hasattr(ctx.composer, "compose_with_sources"):
        try:
            ctx.answer, ctx.answer_type, raw_sources = ctx.composer.compose_with_sources(
                ctx.question,
                ctx.hits,
                memory=ctx.memory,
            )
        except TypeError:
            ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory)
            raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload)
    else:
        ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory)
        raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload)

    ctx.raw_sources = raw_sources
    return AskGraphStep("normalize_sources", "answer composed")


def _ask_graph_normalize_sources(ctx: AskGraphContext) -> AskGraphStep:
    """Normalisiert Quellen und führt finalen Antwort-Postprocessing-Schritt aus.

    Der Orchestratorpfad darf den Composer-Postprocessor nicht umgehen. Deshalb
    werden hier Quellen zuerst nummernstabil normalisiert, danach wird die
    Antwort gegen diese Quellen nachbearbeitet und anschließend erneut leicht
    synchronisiert.
    """
    if ctx.answer_type == "document":
        raw_input_sources = ctx.raw_sources or ctx.sources or []
        ctx.sources = _normalize_returned_sources(
            answer=ctx.answer,
            sources=raw_input_sources,
            hits=ctx.hits,
            payload=ctx.payload,
        )
        ctx.answer = _postprocess_answer(
            ctx.composer,
            ctx.answer,
            sources=ctx.sources,
            hits=ctx.hits or [],
        )
        ctx.sources = _normalize_returned_sources(
            answer=ctx.answer,
            sources=ctx.sources or raw_input_sources,
            hits=ctx.hits,
            payload=ctx.payload,
        )
        # Welche Treffer die Antwort tatsächlich zitiert — noch mit den
        # Retrieval-Rängen, die gleich umnummeriert werden. Danach wäre die
        # Zuordnung verloren: der Text trüge Anzeigenummern, die Treffer weiter
        # ihre Ränge.
        try:
            ctx.cited_hits = AnswerComposer.cited_hits(ctx.answer, ctx.hits or [])
        except Exception as exc:  # noqa: BLE001 - eine Nebenrechnung darf nie die Antwort verhindern.
            logger.warning("Zitatzuordnung übersprungen: %s", exc)
            ctx.cited_hits = []

        # Anzeige-Nummern stabil auf 1..n abbilden (statt Retrieval-Ränge wie
        # "[Quellen 3, 8]"). Muss als letzter Schritt laufen, damit Antworttext
        # und Quellenliste dieselben Nummern zeigen.
        ctx.answer, ctx.sources = _renumber_sources_for_display(ctx.answer, ctx.sources)
    else:
        ctx.sources = []
        ctx.cited_hits = []

    return AskGraphStep("update_memory", "sources normalized and answer postprocessed")


def _ask_graph_update_memory(ctx: AskGraphContext) -> AskGraphStep:
    """Entspricht dem bisherigen memory.add_turn(...)."""
    ctx.memory.add_turn(
        user_message=ctx.question,
        assistant_message=ctx.answer,
        question_kind=classify_question(ctx.question),
    )
    return AskGraphStep("build_response", "conversation memory updated")


_PAGE_RANGE_RE = re.compile(r"(\d+)\s*[–\-]\s*(\d+)|^(\d+)$")


def _attach_pdf_locators(sources: List[Dict[str, Any]], hits: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Sicherheitsnetz: fehlende PDF-Locator aus Hits bzw. page_range ableiten.

    Composer/Orchestrator liefern die Felder normalerweise bereits mit; ältere
    Pfade (Legacy-Fallbacks) können sie verlieren.

    Der Fallback wird pro Dokument gebildet, nicht global: bei mehreren Korpora
    würde "irgendein Hit" sonst das falsche PDF an die Quelle heften und der
    Viewer bei einer SGB-V-Fundstelle den Rahmenvertrag öffnen.
    """
    by_doc: Dict[str, Dict[str, Any]] = {}
    for hit in hits or []:
        metadata = hit.get("metadata") or {}
        doc_id = str(_coalesce(hit.get("doc_id"), metadata.get("doc_id")) or "")
        entry = by_doc.setdefault(doc_id, {})
        entry.setdefault("source_file", _coalesce(hit.get("source_file"), metadata.get("source_file")))
        entry.setdefault("doc_title", _coalesce(hit.get("doc_title"), metadata.get("doc_title")))

    # Nur wenn der Trefferpool eindeutig aus einem Dokument stammt, darf eine
    # Quelle ohne doc_id daraus ergänzt werden.
    single_doc = by_doc.get(next(iter(by_doc))) if len(by_doc) == 1 else None

    for source in sources or []:
        doc_id = str(source.get("doc_id") or "")
        entry = by_doc.get(doc_id) if doc_id else single_doc
        if entry:
            if not source.get("source_file") and entry.get("source_file"):
                source["source_file"] = entry["source_file"]
            if not source.get("doc_title") and entry.get("doc_title"):
                source["doc_title"] = entry["doc_title"]
        if not source.get("doc_id") and single_doc is not None:
            source["doc_id"] = next(iter(by_doc))

        if source.get("page_start") is None:
            match = _PAGE_RANGE_RE.search(str(source.get("page_range") or source.get("pages") or ""))
            if match:
                start = match.group(1) or match.group(3)
                end = match.group(2) or start
                source["page_start"] = _int_or_none(start)
                source["page_end"] = _int_or_none(end)
        if source.get("page_end") is None:
            source["page_end"] = source.get("page_start")

        if not source.get("highlights"):
            source["highlights"] = []

    return sources


def _listenverdikt(
    befund: Any,
    austausch: Any,
    biosimilar: Any = None,
    otc: Any = None,
    verordnung: Any = None,
    lifestyle: Any = None,
    tabak: Any = None,
) -> str:
    """Welcher der Listenbefunde die Kurzantwort setzt.

    Die fünf Listen beantworten teils dieselbe Frageformulierung, aber
    verschiedene Fragen — „dasselbe Präparat eines anderen Herstellers?"
    (Anlage VII Teil B), „eine andere Darreichungsform?" (Teil A), „ein
    Biosimilar statt des Originals?" (Anlage VIIa), „überhaupt zu Lasten der
    GKV?" (Anlagen II, III und I). Die Reihenfolge folgt daraus, wie
    einschneidend die Auskunft ist:

    1. Ein Substitutionsausschluss aus Teil B ist eine harte Schranke und steht
       über allem anderen.
    2. Dann Anlage VIIa, wenn sie entschieden hat: ihre Wirkstoffe sind
       biotechnologisch hergestellt und kommen in keiner der anderen Listen vor
       (geprüft: die Wirkstoffmengen sind disjunkt). Wer nach Humira und
       Amgevita fragt, bekommt sonst „Adalimumab steht nicht auf der
       Substitutionsausschlussliste" — richtig, aber nicht die Frage.
    3. Sonst Teil A, wenn die Frage zwei Formen benannt hat.
    4. Dann Anlage IIa, dann Anlage II, danach Anlage III und erst danach
       Anlage I. Alle vier beantworten dieselbe Frage — „darf das überhaupt zu
       Lasten der GKV verordnet werden?" —, aber sie tragen verschieden weit.

       Anlage IIa steht vor Anlage II, obwohl sie deren Ausnahme ist und nicht
       ihre Regel: ihr Verdikt trägt **beide** Hälften („grundsätzlich
       ausgeschlossen, ausnahmsweise verordnungsfähig, wenn …"), das der Anlage
       II nur eine. Bei Champix stünde sonst „ist ausgeschlossen" als
       Kurzantwort über einem Block, der zwei Absätze tiefer den Anspruch nach
       § 34 Absatz 2 SGB V nennt — genau der Widerspruch, den
       `antwort_mit_befunden` auflösen soll.

       Anlage II steht vor III und I, weil ihr Ausschluss der härteste ist: er
       beruht auf § 34 Absatz 1 Satz 7 SGB V, und der medizinisch begründete
       Einzelfall des § 16 Absatz 5 AM-RL führt an ihm vorbei. Anlage III
       entscheidet auch für verschreibungspflichtige Arzneimittel, hält den
       Einzelfall bei ihren Markern 3 bis 6 aber offen; Anlage I urteilt nur
       über nicht verschreibungspflichtige. Bei Sildenafil steht so „nach
       Anlage II ausgeschlossen" statt des blasseren „steht nicht in Anlage III".
    5. Alle drei stehen hinter den Austauschlisten, weil ihre Signale
       („erstattungsfähig", „Kassenrezept") auch in einer Austauschfrage
       vorkommen, umgekehrt aber nicht.
    6. Erst danach das „nicht gelistet" aus Teil B — die schwächste Aussage von
       allen. Ohne diese Reihenfolge würde eine reine Darreichungsformfrage mit
       „Ambroxol steht nicht auf der Substitutionsausschlussliste" beantwortet:
       richtig, aber am Thema vorbei.
    """
    if befund is not None and befund.status in {
        "ausschluss",
        "ausschluss_bedingt",
        "ausschluss_zwischen_varianten",
    }:
        return amrl_substitution.verdikt(befund)

    for modul, listenbefund in (
        (amrl_biosimilars, biosimilar),
        (amrl_austauschbarkeit, austausch),
        (amrl_tabakentwoehnung, tabak),
        (amrl_lifestyle, lifestyle),
        (amrl_verordnungsausschluss, verordnung),
        (amrl_otc, otc),
    ):
        if listenbefund is not None:
            satz = modul.verdikt(listenbefund)
            if satz:
                return satz

    return amrl_substitution.verdikt(befund) if befund is not None else ""


def _ask_graph_build_response(ctx: AskGraphContext) -> AskGraphStep:
    """Baut exakt die bisherige API-Response-Struktur."""
    _attach_pdf_locators(ctx.sources or [], ctx.hits or [])

    # Korpusgrenze offenlegen: verweist der herangezogene Kontext auf ein
    # Regelwerk, das nicht indiziert ist (AM-RL, AMPreisV, …), wird das benannt.
    # Ohne diesen Hinweis liest sich eine aus § 31 SGB V gebaute Antwort
    # vollständig, obwohl die Verordnungsausschlüsse der AM-RL fehlen.
    #
    # Gelesen werden die *zitierten* Treffer, nicht der ganze Pool. Der Hinweis
    # spricht von den „herangezogenen Textstellen"; im Ausdruck vom 20.08.2026
    # stand unter einer Frage zur Austauschbarkeit von Darreichungsformen der
    # Hinweis auf die BtMVV — sie kommt weder in § 9 Abs. 3 Rahmenvertrag noch
    # in § 40c AM-RL vor, wohl aber in § 6 Rahmenvertrag, den das Modell
    # verworfen hatte. Ein Hinweis auf ein Regelwerk, das die Antwort gar nicht
    # berührt, entwertet die Hinweise, die tragen.
    #
    # Ist keine Zuordnung möglich (Antwort ohne Marker), bleibt es beim Pool:
    # lieber ein Hinweis zu viel als eine verschwiegene Lücke.
    boundary: List[Dict[str, str]] = []
    if ctx.route != "meta":
        try:
            boundary = detect_external_references(
                ctx.cited_hits or ctx.hits or [],
                available_corpora=get_corpus_registry().corpus_ids,
            )
        except Exception as exc:  # noqa: BLE001 - ein Hinweis darf nie die Antwort verhindern.
            logger.warning("Korpusgrenzen-Erkennung übersprungen: %s", exc)
            boundary = []

    # Stand der zitierten Vorschriften: nennt konkret, welche spätere
    # Änderungsvereinbarung eine zitierte Norm geändert hat und was jetzt gilt.
    # Betrifft Korpora, die nur als Grundfassung plus separate Änderungstexte
    # veröffentlicht werden — dort wäre der indizierte Text sonst still veraltet.
    # Beide Hinweise unten hängen an derselben Frage. Sie muss vor dem Abgleich
    # beantwortet werden: reconcile_negative_answer setzt die Fundstelle der
    # Änderungsvereinbarung in den Text und kippte die Prüfung sonst ins
    # Gegenteil.
    nothing_found = _answer_reports_nothing_found(ctx.answer)

    amendments: List[Dict[str, Any]] = []
    amendment_note = ""
    if ctx.route != "meta":
        try:
            # Bewusst die ungefilterten Quellen: ob eine geänderte Vorschrift
            # abgerufen wurde, entscheidet der Trefferbestand, nicht die für die
            # Anzeige gekürzte Liste — die bei einer Verneinung leer ist.
            amendment_note, amendments = describe_amendments(
                ctx.raw_sources or ctx.sources or [],
                path=AMENDMENTS_FILE,
                question=ctx.question,
            )
        except Exception as exc:  # noqa: BLE001 - ein Hinweis darf nie die Antwort verhindern.
            logger.warning("Änderungshinweis übersprungen: %s", exc)
            amendment_note, amendments = "", []

    # Das "nicht geregelt" des Modells gilt nur für den indizierten Stand. Folgt
    # darunter der Änderungshinweis mit der geltenden Regelung, widersprechen
    # sich beide Aussagen im Text, obwohl jede für sich zutrifft. Deshalb wird
    # die Verneinung vor dem Anhängen auf ihren Geltungsbereich zurückgeführt.
    answer = ctx.answer
    if amendments:
        try:
            answer = reconcile_negative_answer(answer, amendments, path=AMENDMENTS_FILE)
        except Exception as exc:  # noqa: BLE001 - siehe oben: nie die Antwort verhindern.
            logger.warning("Abgleich mit dem Änderungshinweis übersprungen: %s", exc)

    # Der Reichweiten-Hinweis spricht von den "herangezogenen Textstellen". Hat
    # die Antwort nichts gefunden, wurde nichts herangezogen: die Verweise
    # stammen dann aus Treffern, die das Modell verworfen hat, und benennen ein
    # fremdes Thema. Beim Beispiel Chargendokumentation waren das AM-RL und BtMVV.
    if boundary and nothing_found:
        logger.info("Reichweiten-Hinweis unterdrückt: Antwort zitiert keine Quelle.")
        boundary = []

    # Substitutionsausschluss: deterministische Prüfung gegen AM-RL Anlage VII
    # Teil B. Die Liste ist bewusst nicht indiziert — über Retrieval wäre die
    # praktisch wichtigste Auskunft ("steht nicht drauf, also austauschbar")
    # nicht belegbar, weil top-k immer Treffer liefert.
    befund = None
    austausch = None
    biosimilar = None
    otc = None
    verordnung = None
    lifestyle = None
    tabak = None
    if ctx.route != "meta":
        # Teil A zuerst: seine Wirkstoffliste ist rund achtmal so lang wie die
        # von Teil B und liefert dessen Negativauskunft den Namen, den die
        # Endungsheuristik allein nicht sicher erkennt.
        try:
            austausch = amrl_austauschbarkeit.pruefe(ctx.question, path=AMRL_AUSTAUSCHBARKEIT_FILE)
        except Exception as exc:  # noqa: BLE001 - eine Prüfung darf nie die Antwort verhindern.
            logger.warning("Austauschbarkeitsprüfung übersprungen: %s", exc)
            austausch = None
        try:
            befund = amrl_substitution.pruefe(
                ctx.question,
                path=AMRL_SUBSTITUTION_FILE,
                kandidat=(austausch.wirkstoff if austausch is not None else None),
            )
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Substitutionsprüfung übersprungen: %s", exc)
            befund = None
        try:
            biosimilar = amrl_biosimilars.pruefe(ctx.question, path=AMRL_BIOSIMILARS_FILE)
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Biosimilar-Prüfung übersprungen: %s", exc)
            biosimilar = None
        try:
            otc = amrl_otc.pruefe(ctx.question, path=AMRL_OTC_FILE)
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("OTC-Prüfung übersprungen: %s", exc)
            otc = None
        try:
            verordnung = amrl_verordnungsausschluss.pruefe(
                ctx.question, path=AMRL_VERORDNUNG_FILE
            )
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Verordnungsausschluss-Prüfung übersprungen: %s", exc)
            verordnung = None
        try:
            lifestyle = amrl_lifestyle.pruefe(ctx.question, path=AMRL_LIFESTYLE_FILE)
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Lifestyle-Prüfung übersprungen: %s", exc)
            lifestyle = None
        try:
            # Der Wirkstoff wird aus Anlage II übergeben, nicht neu erkannt:
            # „Champix" steht dort in der Fertigarzneimittelspalte und wird zu
            # „Vareniclin" aufgelöst, während Anlage IIa gar keine Produkte führt
            # („alle marktverfügbaren Arzneimittel"). Übergeben wird nur, wenn
            # der Treffer die Nikotinabhängigkeit betrifft — sonst beantwortete
            # eine Frage nach Sildenafil eine Frage nach der Tabakentwöhnung.
            tabak = amrl_tabakentwoehnung.pruefe(
                ctx.question,
                path=AMRL_TABAKENTWOEHNUNG_FILE,
                stoff=(
                    lifestyle.treffer[0].wirkstoff
                    if lifestyle is not None and lifestyle.tabakentwoehnung and lifestyle.treffer
                    else None
                ),
            )
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Tabakentwöhnungs-Prüfung übersprungen: %s", exc)
            tabak = None

    geprueft = [
        b
        for b in (befund, austausch, biosimilar, otc, verordnung, lifestyle, tabak)
        if b is not None and b.ist_belastbar
    ]

    # Trägt einer der Befunde eine Auskunft, wurde die Liste geprüft — dann darf
    # der Reichweiten-Hinweis nicht weiter behaupten, sie sei es nicht. Die
    # übrigen Anlagen (Medizinprodukte, Verordnungsausschlüsse …) bleiben
    # ungeprüft, deshalb wird nur der eine Eintrag entfernt und nicht der ganze
    # Hinweis.
    # Dasselbe gilt, wenn das Normregister eine AM-RL-Anlage als Fundstelle
    # benennt: der Hinweis meldete die Anlagen als ungeprüft, während die Antwort
    # sich zwei Absätze darüber auf Anlage VII Teil A berief.
    anlage_benannt = False
    if ctx.route != "meta":
        try:
            anlage_benannt = norm_anchors.anlagen_geprueft(ctx.question)
        except Exception as exc:  # noqa: BLE001 - siehe oben.
            logger.warning("Anlagenprüfung des Normregisters übersprungen: %s", exc)

    if (geprueft or anlage_benannt) and boundary:
        boundary = [ref for ref in boundary if ref.get("key") != "am_rl_anlagen"]

    # Die Listenbefunde stehen ganz oben: sie sind die Auskunft, das Modell
    # darunter ordnet ein, was daraus für Abgabe und Abrechnung folgt. Die
    # Kurzantwort wird dabei angeglichen — ohne die Listen kommt das Modell zur
    # allgemeinen Ersetzungsregel und widerspricht dem Befund zwei Zeilen darüber.
    if geprueft:
        answer = amrl_substitution.antwort_mit_befunden(
            answer,
            bloecke=[
                amrl_substitution.befund_block(befund) if befund is not None else "",
                amrl_austauschbarkeit.befund_block(austausch) if austausch is not None else "",
                amrl_biosimilars.befund_block(biosimilar) if biosimilar is not None else "",
                amrl_lifestyle.befund_block(lifestyle) if lifestyle is not None else "",
                amrl_tabakentwoehnung.befund_block(tabak) if tabak is not None else "",
                amrl_verordnungsausschluss.befund_block(verordnung)
                if verordnung is not None
                else "",
                amrl_otc.befund_block(otc) if otc is not None else "",
            ],
            verdikt=_listenverdikt(
                befund, austausch, biosimilar, otc, verordnung, lifestyle, tabak
            ),
        )

    # Normebene statt Stoffebene: die vier Listen prüfen, ob ein Wirkstoff auf
    # einer Anlage steht, und melden bei einer reinen Normfrage planmäßig
    # `nicht_pruefbar`. Genau dort lagen die beiden schwersten Fehler der
    # Auswertung — eine OTC-Frage mit § 31 statt § 34 beantwortet, das
    # Zuweisungsverbot aus § 360 hergeleitet statt aus § 31 Abs. 1 Satz 6.
    normbefund = None
    if ctx.route != "meta":
        try:
            normbefund = norm_anchors.pruefe(ctx.question, answer)
        except Exception as exc:  # noqa: BLE001 - ein Abgleich darf nie die Antwort verhindern.
            logger.warning("Normabgleich übersprungen: %s", exc)
            normbefund = None

    if normbefund is not None:
        # Steht die Antwort auf der falschen Vorschrift, werden „Kurzantwort" und
        # „Maßgebliche Norm" neu gesetzt — ein bloßer Nachtrag stünde neben der
        # falschen Antwort, und nichts im Text sagte, welcher gilt. Fehlt nur ein
        # Glied der Kette, bleibt die Antwort stehen und bekommt den Nachtrag.
        answer = norm_anchors.in_antwort_einsetzen(answer, normbefund)

    # Beide Hinweise erst jetzt: sie werden ans Ende gehängt, und dort steht
    # ohne eigene Überschrift noch der Abschnitt „Kurzantwort". Setzt ein
    # Listenbefund diese Kurzantwort, ersetzt `antwort_mit_befunden` den ganzen
    # Abschnitt — ein vorher angehängter Hinweis steckte darin und wäre spurlos
    # verschwunden. Die Reihenfolge der beiden zueinander bleibt: der
    # Änderungshinweis trägt die eigentliche Auskunft und steht deshalb vor dem
    # Reichweiten-Hinweis, der nur eine Einschränkung benennt.
    answer = append_amendment_note(answer, amendment_note)
    if boundary:
        answer = append_boundary_note(answer, boundary)

    ctx.response_body = {
        "question": ctx.question,
        "answer": answer,
        "answer_type": ctx.answer_type,
        "session_id": ctx.request.state.session_id,
        "factual_question_index": ctx.memory.factual_question_count,
        "total_turns": ctx.memory.total_turns,
        "sources": ctx.sources or [],
        "needs_clarification": ctx.needs_clarification,
        "clarification_question": ctx.clarification_question,
        # Maschinenlesbar für UI/Monitoring: welche Regelwerke fehlten.
        "corpus_boundary": boundary,
        # Zitierte Vorschriften, deren indizierter Stand überholt ist.
        "corpus_amendments": [
            {"locator": p.get("locator"), "status": p.get("status"), "current_rule": p.get("current_rule")}
            for p in amendments
        ],
        # Welche Norm das Register für maßgeblich hält und ob die Antwort sie nennt.
        "norm_anchor": normbefund.to_dict() if normbefund is not None else None,
        # Maschinenlesbare Listenbefunde samt Stand — für UI und Monitoring.
        "substitutionsausschluss": befund.to_dict() if befund is not None else None,
        "austauschbarkeit": austausch.to_dict() if austausch is not None else None,
        "biosimilars": biosimilar.to_dict() if biosimilar is not None else None,
        "otc_verordnungsfaehigkeit": otc.to_dict() if otc is not None else None,
        "verordnungsausschluss": verordnung.to_dict() if verordnung is not None else None,
        "lifestyle_ausschluss": lifestyle.to_dict() if lifestyle is not None else None,
        "tabakentwoehnung": tabak.to_dict() if tabak is not None else None,
    }
    return AskGraphStep("__end__", "response built")


ASK_GRAPH_NODES: Dict[AskNodeName, Callable[[AskGraphContext], AskGraphStep]] = {
    "meta_decision": _ask_graph_meta_decision,
    "retrieve_hits": _ask_graph_retrieve_hits,
    "compose_answer": _ask_graph_compose_answer,
    "orchestrate_answer": _ask_graph_orchestrate_answer,
    "normalize_sources": _ask_graph_normalize_sources,
    "update_memory": _ask_graph_update_memory,
    "build_response": _ask_graph_build_response,
}

ASK_GRAPH: Dict[str, List[Dict[str, str]]] = {
    "__start__": [{"to": "meta_decision", "label": ""}],
    "meta_decision": [
        {"to": "compose_answer", "label": "meta"},
        {"to": "orchestrate_answer", "label": "non_meta"},
        {"to": "retrieve_hits", "label": "legacy"},
    ],
    "retrieve_hits": [{"to": "compose_answer", "label": ""}],
    "orchestrate_answer": [{"to": "normalize_sources", "label": ""}],
    "compose_answer": [{"to": "normalize_sources", "label": ""}],
    "normalize_sources": [{"to": "update_memory", "label": ""}],
    "update_memory": [{"to": "build_response", "label": ""}],
    "build_response": [{"to": "__end__", "label": ""}],
}


class AskGraphVizState(TypedDict, total=False):
    """Minimaler LangGraph-State nur für Visualisierung/Rendering."""

    route: str


def _ask_langgraph_passthrough(state: AskGraphVizState) -> AskGraphVizState:
    """Dummy-Node für LangGraph-Rendering; die echte Logik bleibt in run_ask_graph."""
    return state


def _make_ask_langgraph_router(source: str, labels: List[str]):
    """
    Erzeugt einen Router für LangGraph-Visualisierung.

    Für das Rendering ist nur die Mapping-Struktur wichtig. Falls der Graph
    doch testweise ausgeführt wird, kann pro Node über '<node>_route' geroutet
    werden; sonst wird der erste Label-Zweig genutzt.
    """
    default_label = labels[0]

    def _router(state: AskGraphVizState) -> str:
        return state.get(f"{source}_route", state.get("route", default_label))

    return _router


def build_ask_workflow():
    """
    Baut den LangGraph-Workflow aus ASK_GRAPH.

    Diese Funktion ist die stabile Schnittstelle für visualize_graph.py:
        from app import build_ask_workflow
        workflow = build_ask_workflow()
        workflow.get_graph(xray=True).draw_mermaid_png()

    Wenn ASK_GRAPH in app.py geändert wird, übernimmt die Visualisierung diese
    Änderung automatisch, ohne dass visualize_graph.py angepasst werden muss.
    """
    if StateGraph is None:
        raise RuntimeError(
            "LangGraph ist nicht installiert. Bitte ausführen: pip install -U langgraph langchain-core"
        )

    workflow = StateGraph(AskGraphVizState)

    node_names = set()
    for source, edges in ASK_GRAPH.items():
        if source not in {"__start__", "__end__"}:
            node_names.add(source)
        for edge in edges:
            target = edge["to"]
            if target not in {"__start__", "__end__"}:
                node_names.add(target)

    for node_name in sorted(node_names):
        workflow.add_node(node_name, _ask_langgraph_passthrough)

    start_edges = ASK_GRAPH.get("__start__") or []
    if not start_edges:
        raise RuntimeError("ASK_GRAPH benötigt eine __start__-Kante.")
    workflow.set_entry_point(start_edges[0]["to"])

    for source, edges in ASK_GRAPH.items():
        if source in {"__start__", "__end__"}:
            continue

        labelled_edges = [edge for edge in edges if edge.get("label")]
        plain_edges = [edge for edge in edges if not edge.get("label")]

        if labelled_edges:
            labels = [edge["label"] for edge in labelled_edges]
            workflow.add_conditional_edges(
                source,
                _make_ask_langgraph_router(source, labels),
                {edge["label"]: (END if edge["to"] == "__end__" else edge["to"]) for edge in labelled_edges},
            )

        for edge in plain_edges:
            target = END if edge["to"] == "__end__" else edge["to"]
            workflow.add_edge(source, target)

    return workflow.compile()


def get_ask_workflow():
    """Alias für Visualisierungsskripte, die eine Getter-Funktion bevorzugen."""
    return build_ask_workflow()


# Backwards-/Convenience-Aliase für Visualisierungsskripte.
# Wichtig: FastAPI bleibt weiterhin in der Variable `app`; der LangGraph darf
# deshalb nicht ebenfalls `app` heißen.
try:
    ASK_WORKFLOW = build_ask_workflow()
except Exception:
    ASK_WORKFLOW = None

AGENT_GRAPH = ASK_GRAPH
ASK_LANGGRAPH_APP = ASK_WORKFLOW


def _ask_graph_mermaid() -> str:
    lines = ["flowchart TD"]
    for source, edges in ASK_GRAPH.items():
        for edge in edges:
            target = edge["to"]
            label = edge.get("label") or ""
            if label:
                lines.append(f'    {source}["{source}"] -- "{label}" --> {target}["{target}"]')
            else:
                lines.append(f'    {source}["{source}"] --> {target}["{target}"]')
    return "\n".join(lines)


def run_ask_graph(ctx: AskGraphContext) -> AskGraphContext:
    """Führt den /ask-Graphen aus, ohne fachliche Zusatzpfade einzuführen."""
    current: AskNodeName = "meta_decision"

    for _ in range(MAX_ASK_GRAPH_STEPS):
        node = ASK_GRAPH_NODES.get(current)
        if node is None:
            raise HTTPException(status_code=500, detail=f"Unbekannter Ask-Graph-Knoten: {current}")

        step = node(ctx)
        ctx.trace.append(
            AskGraphTraceEntry(
                node=current,
                next_node=step.next_node,
                reason=step.reason,
                route=ctx.route,
                hit_count=len(ctx.hits or []),
                answer_type=ctx.answer_type,
            )
        )

        if step.next_node == "__end__":
            return ctx

        current = step.next_node

    raise HTTPException(
        status_code=500,
        detail=f"Ask-Graph nach {MAX_ASK_GRAPH_STEPS} Schritten abgebrochen.",
    )


def _legacy_answer_flow(ctx: AskGraphContext) -> AskGraphContext:
    """
    Fallback auf die bisherige monolithische /ask-Logik.

    Dieser Pfad bleibt absichtlich funktionsgleich zum Graphen und dient nur
    als schneller Rollback über ASK_GRAPH_ENABLED=false.
    """
    if is_meta_question(ctx.question):
        ctx.route = "meta"
        ctx.hits = []
        ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, [], memory=ctx.memory)
        ctx.sources = []
    else:
        ctx.route = "non_meta"
        ctx.hits = _retrieve_hits(ctx.payload)

        if hasattr(ctx.composer, "compose_with_sources"):
            try:
                ctx.answer, ctx.answer_type, raw_sources = ctx.composer.compose_with_sources(
                    ctx.question,
                    ctx.hits,
                    memory=ctx.memory,
                )
            except TypeError:
                ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory)
                raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload)
        else:
            ctx.answer, ctx.answer_type = ctx.composer.compose(ctx.question, ctx.hits, memory=ctx.memory)
            raw_sources = _build_sources_from_composer_or_hits(ctx.composer, ctx.hits, ctx.payload)

        ctx.raw_sources = raw_sources
        if ctx.answer_type == "document":
            ctx.sources = _normalize_returned_sources(
                answer=ctx.answer,
                sources=raw_sources,
                hits=ctx.hits,
                payload=ctx.payload,
            )
            ctx.answer = _postprocess_answer(
                ctx.composer,
                ctx.answer,
                sources=ctx.sources,
                hits=ctx.hits or [],
            )
            ctx.sources = _normalize_returned_sources(
                answer=ctx.answer,
                sources=ctx.sources or raw_sources,
                hits=ctx.hits,
                payload=ctx.payload,
            )
        else:
            ctx.sources = []

    ctx.memory.add_turn(
        user_message=ctx.question,
        assistant_message=ctx.answer,
        question_kind=classify_question(ctx.question),
    )
    _ask_graph_build_response(ctx)
    return ctx


def _ask_graph_debug_payload(ctx: AskGraphContext) -> Dict[str, Any]:
    return {
        "ask_graph_enabled": ASK_GRAPH_ENABLED,
        "orchestrator_enabled": ORCHESTRATOR_ENABLED,
        "orchestrator": ctx.orchestrator_debug,
        "workflow_trace": [entry.__dict__ for entry in (ctx.trace or [])],
        "workflow_graph": ASK_GRAPH,
        "workflow_mermaid": _ask_graph_mermaid(),
        "route": ctx.route,
        "is_meta_question": is_meta_question(ctx.question),
        "question_kind": classify_question(ctx.question),
        "retrieved_hit_count": len(ctx.hits or []),
        "answer_source_numbers": _extract_answer_source_numbers(ctx.answer),
        "raw_sources": ctx.raw_sources or [],
        "normalized_sources": ctx.sources or [],
        "hits": [_debug_hit(hit) for hit in (ctx.hits or [])],
        "top_k": ctx.payload.top_k or DEFAULT_TOP_K,
        "fetch_k": ctx.payload.fetch_k or DEFAULT_FETCH_K,
        "max_final_results": ctx.payload.max_final_results or DEFAULT_MAX_FINAL_RESULTS,
        "max_sources": ctx.payload.max_sources or DEFAULT_MAX_SOURCES,
        "include_neighbors": ctx.payload.include_neighbors,
        "include_explicit_sections": ctx.payload.include_explicit_sections,
        "restrict_to_default_container": ctx.payload.restrict_to_default_container,
        "source_sync_to_answer_markers": SOURCE_SYNC_TO_ANSWER_MARKERS,
        "chroma_persist_dir": CHROMA_PERSIST_DIR,
        "chroma_collection": CHROMA_COLLECTION,
        "chroma_count": _collection_count(),
        "corpus_counts": _corpus_counts(),
        "routed_corpora": sorted({
            str(hit.get("corpus_id")) for hit in (ctx.hits or []) if hit.get("corpus_id")
        }),
    }


# -----------------------------------------------------------------------------
# API Endpoints
# -----------------------------------------------------------------------------

@app.post("/ask")
def ask(payload: Question, request: Request):
    question = _clean_question(payload.question)
    if not question:
        raise HTTPException(status_code=422, detail="question darf nicht leer sein.")

    session: SessionState = request.state.session
    memory = session.get_memory()

    llm = _build_llm(session)
    composer = _build_composer(llm, payload)
    orchestrator = _build_orchestrator(get_retriever(), composer, payload) if ORCHESTRATOR_ENABLED else None

    ctx = AskGraphContext(
        payload=payload,
        request=request,
        session=session,
        memory=memory,
        llm=llm,
        composer=composer,
        question=question,
        orchestrator=orchestrator,
    )

    ctx = run_ask_graph(ctx) if ASK_GRAPH_ENABLED else _legacy_answer_flow(ctx)

    response_body: Dict[str, Any] = ctx.response_body or {
        "question": question,
        "answer": ctx.answer,
        "answer_type": ctx.answer_type,
        "session_id": request.state.session_id,
        "factual_question_index": memory.factual_question_count,
        "total_turns": memory.total_turns,
        "sources": ctx.sources or [],
        "needs_clarification": ctx.needs_clarification,
        "clarification_question": ctx.clarification_question,
        "corpus_boundary": [],
        # Der Fallback greift nur, wenn der Graph keinen Body gebaut hat; die
        # Schlüssel bleiben trotzdem stabil, damit die UI nicht zwei Formen kennen muss.
        "substitutionsausschluss": None,
        "austauschbarkeit": None,
        "biosimilars": None,
        "otc_verordnungsfaehigkeit": None,
        "verordnungsausschluss": None,
        "lifestyle_ausschluss": None,
        "tabakentwoehnung": None,
    }

    if payload.debug:
        response_body["debug"] = _ask_graph_debug_payload(ctx)

    return response_body


@app.get("/pdf/{filename}")
def get_pdf(filename: str):
    """Liefert ein Quell-PDF für den eingebetteten Fundstellen-Viewer aus.

    Es werden nur einfache PDF-Dateinamen akzeptiert (kein Pfadanteil), und
    die Datei muss in einem der konfigurierten PDF-Verzeichnisse liegen.
    """
    name = os.path.basename(filename or "").strip()
    if not name or not _PDF_NAME_RE.match(name):
        raise HTTPException(status_code=400, detail="Ungültiger PDF-Dateiname.")

    for directory in PDF_SEARCH_DIRS:
        candidate = (directory / name).resolve()
        try:
            candidate.relative_to(directory.resolve())
        except (ValueError, OSError):
            continue
        if candidate.is_file():
            return FileResponse(
                str(candidate),
                media_type="application/pdf",
                headers={
                    "Content-Disposition": f'inline; filename="{name}"',
                    "Cache-Control": "public, max-age=3600",
                },
            )

    raise HTTPException(status_code=404, detail=f"PDF nicht gefunden: {name}")


@app.get("/debug/workflow")
def debug_workflow():
    return {
        "ok": True,
        "ask_graph_enabled": ASK_GRAPH_ENABLED,
        "max_ask_graph_steps": MAX_ASK_GRAPH_STEPS,
        "langgraph_available": StateGraph is not None,
        "graph": ASK_GRAPH,
        "mermaid": _ask_graph_mermaid(),
    }


@app.get("/debug/workflow/mermaid")
def debug_workflow_mermaid():
    return Response(_ask_graph_mermaid(), media_type="text/plain")


@app.get("/health")
def health():
    return {
        "ok": True,
        "app": APP_TITLE,
        "version": APP_VERSION,
        "collection": CHROMA_COLLECTION,
        "corpora": _corpus_counts(),
        "chroma_persist_dir": CHROMA_PERSIST_DIR,
        "chroma_count": _collection_count(),
        "default_container_id": DEFAULT_CONTAINER_ID,
        "embedding_model": EMBEDDING_MODEL,
        "groq_model": GROQ_MODEL,
        "source_sync_to_answer_markers": SOURCE_SYNC_TO_ANSWER_MARKERS,
        "ask_graph_enabled": ASK_GRAPH_ENABLED,
        "orchestrator_enabled": ORCHESTRATOR_ENABLED,
    }


@app.get("/debug/retriever")
def debug_retriever():
    """Diagnose pro Korpus statt für eine einzelne Collection."""
    try:
        registry = get_corpus_registry()
        diagnostics = registry.diagnostics()
    except Exception as exc:
        return {
            "ok": False,
            "error": f"{type(exc).__name__}: {exc}",
            "chroma_persist_dir": CHROMA_PERSIST_DIR,
        }

    for entry in diagnostics["corpora"]:
        if not entry.get("ok"):
            continue
        try:
            res = registry.retriever(entry["corpus_id"]).col.get(limit=3, include=["metadatas", "documents"])
            docs = res.get("documents") or []
            metas = res.get("metadatas") or []
            ids = res.get("ids") or []
            entry["sample"] = [
                {
                    "id": ids[idx] if idx < len(ids) else None,
                    "metadata": meta,
                    "normalized_source": _source_from_hit({"metadata": meta}),
                    "text_preview": (docs[idx] if idx < len(docs) else "")[:250],
                }
                for idx, meta in enumerate(metas)
            ]
        except Exception as exc:  # noqa: BLE001
            entry["sample_error"] = f"{type(exc).__name__}: {exc}"

    return {
        "ok": True,
        "chroma_persist_dir": CHROMA_PERSIST_DIR,
        "count": _collection_count(),
        "embedding_consistency": registry.assert_consistent_embedding() or "ok",
        **diagnostics,
    }


@app.get("/debug/routing")
def debug_routing(question: str = ""):
    """Zeigt, welche Korpora eine Frage erreichen würde — und warum."""
    try:
        available = get_corpus_registry().available()
    except Exception as exc:  # noqa: BLE001
        return {"ok": False, "error": f"{type(exc).__name__}: {exc}"}

    out: Dict[str, Any] = {"ok": True, **explain_routing(question, available)}

    # Der Router sieht nur den Wortlaut. Ob ein Korpus zusätzlich zwingend
    # abgefragt wird, entscheidet das Normregister — und genau dieser Fall ist
    # der, den man debuggen will ("Sonderkennzeichen" nennt kein Korpus).
    try:
        out["federation"] = get_retriever().explain_selection(question)
    except Exception as exc:  # noqa: BLE001
        out["federation_error"] = f"{type(exc).__name__}: {exc}"

    return out


# -----------------------------------------------------------------------------
# Session System-Prompt
# -----------------------------------------------------------------------------

@app.get("/system-prompt")
def get_system_prompt(request: Request):
    session: SessionState = request.state.session
    return {
        "system_prompt": session.system_prompt,
        "default_system_prompt": DEFAULT_SYSTEM_PROMPT,
    }


@app.post("/system-prompt")
def set_system_prompt(payload: SystemPromptPayload, request: Request):
    prompt = payload.system_prompt.strip()
    if not prompt:
        raise HTTPException(status_code=422, detail="system_prompt darf nicht leer sein.")

    request.state.session.system_prompt = prompt
    return {
        "ok": True,
        "system_prompt": request.state.session.system_prompt,
    }


@app.post("/system-prompt/reset")
def reset_system_prompt(request: Request):
    request.state.session.system_prompt = DEFAULT_SYSTEM_PROMPT
    return {
        "ok": True,
        "system_prompt": request.state.session.system_prompt,
    }


# -----------------------------------------------------------------------------
# Session zurücksetzen und Verlauf abrufen
# -----------------------------------------------------------------------------

@app.post("/session/reset")
def reset_session(request: Request):
    session: SessionState = request.state.session
    session.reset()
    return {"ok": True}


@app.get("/session/history")
def get_history(request: Request):
    session: SessionState = request.state.session
    memory = session.get_memory()
    return {
        "factual_questions": memory.get_factual_qa(),
        "total_turns": memory.total_turns,
        "factual_question_count": memory.factual_question_count,
    }


@app.get("/session")
def get_session_info(request: Request):
    session: SessionState = request.state.session
    memory = session.get_memory()
    return {
        "session_id": request.state.session_id,
        "total_turns": memory.total_turns,
        "factual_question_count": memory.factual_question_count,
    }


# -----------------------------------------------------------------------------
# UI
# -----------------------------------------------------------------------------

@app.get("/")
def index():
    if not os.path.exists(INDEX_FILE):
        raise HTTPException(
            status_code=404,
            detail=f"UI-Datei nicht gefunden: {INDEX_FILE}",
        )
    return FileResponse(INDEX_FILE)