feat(agent): harden API-first service

pv-agent/agent/README.md

@@ -43,9 +43,14 @@ python -m agent.cli eval
 python -m agent.cli eval --answers --json-out data/eval-report.json
 
 # 4) HTTP-API + Test-Chat
-python -m agent.cli serve             # http://127.0.0.1:8080 (/ask, /health, /reindex)
+python -m agent.cli serve             # http://127.0.0.1:8080 (/v1/ask, /v1/health, /v1/reindex)
 ```
 
+Der stabile v1-Vertrag, Bearer-Authentisierung, Fehlersemantik und die
+Datenschutzgrenze für die spätere Odoo-Anbindung sind in `docs/API.md`
+dokumentiert. v1 ist zustandslos und akzeptiert keine Mandanten-, Mitarbeiter-
+oder Abrechnungsobjekte.
+
 ## Konfiguration (Umgebungsvariablen)
 
 | Variable | Default | Bedeutung |
@@ -69,6 +74,8 @@ python -m agent.cli serve             # http://127.0.0.1:8080 (/ask, /health, /r
 | `PV_MAX_CONTEXT_CHARS` | `90000` | User-Content-Budget; niedrig gerankte Blöcke werden ganz weggelassen (`trim_results`) |
 | `PV_CONTEXT_BLOCKS` | `8` | Kontextblöcke im Prompt |
 | `PV_PORT` | `8080` | API-Port |
+| `PV_API_KEY` | leer | Bearer-Key für `/v1/ask`; leer nur für lokale Entwicklung ohne Auth |
+| `PV_ADMIN_API_KEY` | leer | separater Bearer-Key für `/v1/reindex`; leer = `PV_API_KEY` verwenden |
 
 ## Deployment auf dem Host (Ollama-Maschine)
 
@@ -206,7 +213,7 @@ agent/
   retrieve.py       Hybrid-Retrieval (BM25 + Dense, RRF, cross_refs)
   ollama_client.py  Ollama-HTTP (embed + chat, think-Fallback)
   generate.py       Systemprompt, Post-Validierung, Verweigerung
-  api.py            FastAPI (/ask, /health, /reindex)
+  api.py            FastAPI v1 (/v1/ask, /v1/health, /v1/reindex), Auth + Request-IDs
   cli.py            ingest | ask | eval | serve
   eval/             goldset.yaml + evaluate.py
 web/index.html      Minimaler Test-Chat

pv-agent/agent/api.py

@@ -1,31 +1,51 @@
-"""FastAPI-Oberfläche des PV RAG Agent.
+"""Versionierte FastAPI-Oberfläche des PV RAG Agent.
 
-Endpunkte:
-  POST /ask      — Frage -> belegte Antwort (oder Verweigerung)
-  GET  /health   — Index- und Ollama-Status
-  POST /reindex  — Index-Neuaufbau (nach neuem Wissensbasis-Batch)
+Die API bleibt ein reiner Wissensdienst: Requests enthalten eine Frage, aber
+keinen Mandanten- oder Payroll-Datenkontext. Eine spätere Lohndaten-Anbindung
+benötigt einen getrennten, mandantenautorisierten Vertrag.
 """
 from __future__ import annotations
 
+import logging
+import re
+import secrets
 import threading
+import uuid
 from contextlib import asynccontextmanager
+from typing import Literal
 
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel, Field
+from fastapi import Depends, FastAPI, HTTPException, Request
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from pydantic import BaseModel, ConfigDict, Field
 
 from .config import Config
-from .generate import answer_question
+from .generate import CITE_RE, answer_question
 from .ingest import build_index
 from .ollama_client import OllamaClient
 from .retrieve import Retriever
 
+API_VERSION = "v1"
+DATA_SCOPE = "knowledge_base_only"
+_REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
+logger = logging.getLogger(__name__)
+bearer = HTTPBearer(auto_error=False)
+bearer_credentials = Depends(bearer)
 
-class AskRequest(BaseModel):
+
+class StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class AskRequest(StrictModel):
     question: str = Field(min_length=3, max_length=2000)
     top_k: int | None = Field(default=None, ge=1, le=20)
+    mode: Literal["knowledge"] = Field(
+        default="knowledge",
+        description="Derzeit ausschließlich KB-Wissen; kein Payroll-Datenkontext.",
+    )
 
 
-class SourceOut(BaseModel):
+class SourceOut(StrictModel):
     id: str
     title: str
     section: str | None = None
@@ -33,51 +53,93 @@ class SourceOut(BaseModel):
     work: str | None = None
 
 
-class AskResponse(BaseModel):
+class ConflictOut(StrictModel):
+    summary: str
+    source_ids: list[str]
+
+
+class PlannedQueryOut(StrictModel):
+    text: str
+    stand_year: str | None = None
+    scope: str | None = None
+
+
+class GroundingOut(StrictModel):
+    data_scope: Literal["knowledge_base_only"] = DATA_SCOPE
+    citations_verified: bool
+    context_count: int
+    regenerations: int
+
+
+class AskResponse(StrictModel):
+    api_version: Literal["v1"] = API_VERSION
+    request_id: str
+    status: Literal["answered", "refused", "uncertain"]
     question: str
     answer: str
     refused: bool
     verified: bool
     citations: list[str]
     sources: list[SourceOut]
+    conflicts: list[ConflictOut] = Field(default_factory=list)
+    assumptions: list[str] = Field(default_factory=list)
+    clarification_question: str | None = None
+    alternatives: list[str] = Field(default_factory=list)
+    answer_type: Literal["specific", "survey"] = "specific"
+    planned: bool = False
+    planned_queries: list[PlannedQueryOut] = Field(default_factory=list)
+    grounding: GroundingOut
     n_context: int
     model: str
     latency_ms: int
     regenerations: int = 0
 
 
+class HealthResponse(StrictModel):
+    api_version: Literal["v1"] = API_VERSION
+    service: Literal["pv-rag-agent"] = "pv-rag-agent"
+    status: Literal["ok", "degraded"]
+    index: dict
+    ollama_up: bool
+    authentication_enabled: bool
+
+
 class AppState:
     def __init__(self) -> None:
         self.cfg: Config | None = None
         self.client: OllamaClient | None = None
         self.retriever: Retriever | None = None
-        self.lock = threading.Lock()
+        self.maintenance_lock = threading.Lock()
+        self.state_lock = threading.RLock()
 
     def ensure(self) -> Config:
-        if self.cfg is None:
-            self.cfg = Config.from_env()
-        return self.cfg
+        with self.state_lock:
+            if self.cfg is None:
+                self.cfg = Config.from_env()
+            return self.cfg
 
     def get_client(self) -> OllamaClient:
-        if self.client is None:
-            cfg = self.ensure()
-            self.client = OllamaClient(
-                cfg.ollama_url,
-                embed_timeout_s=cfg.embed_timeout_s,
-                chat_timeout_s=cfg.chat_timeout_s,
-            )
-        return self.client
+        with self.state_lock:
+            if self.client is None:
+                cfg = self.ensure()
+                self.client = OllamaClient(
+                    cfg.ollama_url,
+                    embed_timeout_s=cfg.embed_timeout_s,
+                    chat_timeout_s=cfg.chat_timeout_s,
+                )
+            return self.client
 
     def get_retriever(self) -> Retriever:
-        if self.retriever is None:
-            cfg = self.ensure()
-            self.retriever = Retriever(cfg)
-        return self.retriever
+        with self.state_lock:
+            if self.retriever is None:
+                self.retriever = Retriever(self.ensure())
+            return self.retriever
 
     def reset_retriever(self) -> None:
-        if self.retriever is not None:
-            self.retriever.close()
-        self.retriever = None
+        with self.state_lock:
+            if self.retriever is not None:
+                self.retriever.close()
+            self.retriever = None
 
 
 @asynccontextmanager
@@ -85,72 +147,233 @@ async def lifespan(app: FastAPI):
     app.state.rag = AppState()
     yield
     rag: AppState = app.state.rag
-    if rag.retriever:
-        rag.retriever.close()
+    rag.reset_retriever()
     if rag.client:
         rag.client.close()
 
 
-app = FastAPI(title="PV RAG Agent", version="0.1.0", lifespan=lifespan)
+app = FastAPI(
+    title="PV RAG Agent",
+    version="1.0.0",
+    description=(
+        "Eigenständiger, KB-gebundener Wissensdienst. Der v1-Vertrag nimmt "
+        "keine Mandanten- oder Mitarbeiterdaten entgegen."
+    ),
+    lifespan=lifespan,
+)
 
 
-@app.post("/ask", response_model=AskResponse)
-def ask(req: AskRequest) -> AskResponse:
-    rag: AppState = app.state.rag
+@app.middleware("http")
+async def add_request_id(request: Request, call_next):
+    supplied = request.headers.get("X-Request-ID", "")
+    request_id = supplied if _REQUEST_ID_RE.fullmatch(supplied) else uuid.uuid4().hex
+    request.state.request_id = request_id
+    response = await call_next(request)
+    response.headers["X-Request-ID"] = request_id
+    return response
+
+
+def _request_id(request: Request) -> str:
+    return getattr(request.state, "request_id", uuid.uuid4().hex)
+
+
+def _check_bearer(
+    expected: str,
+    credentials: HTTPAuthorizationCredentials | None,
+) -> None:
+    # Leere Keys erhalten den bisherigen lokalen Entwicklungsmodus. Für ein
+    # exponiertes Deployment muss PV_API_KEY gesetzt sein.
+    if not expected:
+        return
+    supplied = credentials.credentials if credentials and credentials.scheme.lower() == "bearer" else ""
+    if not supplied or not secrets.compare_digest(supplied, expected):
+        raise HTTPException(
+            status_code=401,
+            detail="Authentisierung erforderlich.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+def require_api_access(
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = bearer_credentials,
+) -> None:
+    rag: AppState = request.app.state.rag
+    _check_bearer(rag.ensure().api_key, credentials)
+
+
+def require_admin_access(
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = bearer_credentials,
+) -> None:
+    rag: AppState = request.app.state.rag
     cfg = rag.ensure()
+    _check_bearer(cfg.admin_api_key or cfg.api_key, credentials)
+
+
+def _extract_conflicts(answer: str, citations: list[str]) -> list[ConflictOut]:
+    allowed = set(citations)
+    conflicts: list[ConflictOut] = []
+    for paragraph in re.split(r"\n\s*\n", answer):
+        text = " ".join(paragraph.split())
+        if "⚠" not in text:
+            continue
+        ids = sorted(set(CITE_RE.findall(text)) & allowed)
+        if ids:
+            conflicts.append(ConflictOut(summary=text, source_ids=ids))
+    return conflicts
+
+
+def _extract_clarification(answer: str, refused: bool) -> str | None:
+    if refused:
+        return None
+    match = re.search(r"([^.!?\n]*\?)\s*$", answer.strip())
+    if not match:
+        return None
+    question = match.group(1).strip().lstrip("-• ")
+    return question or None
+
+
+def _response_from_result(result: dict, request_id: str) -> AskResponse:
+    if not result["verified"]:
+        status = "uncertain"
+    elif result["refused"]:
+        status = "refused"
+    else:
+        status = "answered"
+    citations = list(result["citations"])
+    return AskResponse(
+        request_id=request_id,
+        status=status,
+        question=result["question"],
+        answer=result["answer"],
+        refused=result["refused"],
+        verified=result["verified"],
+        citations=citations,
+        sources=result["sources"],
+        conflicts=_extract_conflicts(result["answer"], citations),
+        # Keine Annahmen oder Alternativen aus Freitext erraten. Diese Felder
+        # sind Teil des stabilen Vertrags und werden erst befüllt, wenn die
+        # Generierung sie selbst belegbar strukturiert liefert.
+        assumptions=[],
+        clarification_question=_extract_clarification(
+            result["answer"], result["refused"]
+        ),
+        alternatives=[],
+        answer_type=result.get("answer_type", "specific"),
+        planned=result.get("planned", False),
+        planned_queries=result.get("planned_queries", []),
+        grounding=GroundingOut(
+            citations_verified=result["verified"],
+            context_count=result["n_context"],
+            regenerations=result["regenerations"],
+        ),
+        n_context=result["n_context"],
+        model=result["model"],
+        latency_ms=result["latency_ms"],
+        regenerations=result["regenerations"],
+    )
+
+
+def _ask(req: AskRequest, request: Request) -> AskResponse:
+    rag: AppState = request.app.state.rag
+    cfg = rag.ensure()
+    request_id = _request_id(request)
     try:
         result = answer_question(
-            req.question, cfg,
+            req.question,
+            cfg,
             client=rag.get_client(),
             retriever=rag.get_retriever(),
             top_k=req.top_k,
         )
-    except RuntimeError as e:  # Index fehlt
-        raise HTTPException(status_code=503, detail=str(e)) from e
-    except Exception as e:  # Ollama nicht erreichbar o. Ä.
+    except Exception as exc:
+        logger.exception("Antwortgenerierung fehlgeschlagen request_id=%s", request_id)
         raise HTTPException(
             status_code=503,
-            detail=f"Antwortgenerierung fehlgeschlagen: {type(e).__name__}: {e}",
-        ) from e
+            detail="Der Wissensdienst ist vorübergehend nicht verfügbar.",
+        ) from exc
     result.pop("draft", None)
-    return AskResponse(**result)
+    return _response_from_result(result, request_id)
 
 
-@app.get("/health")
-def health() -> dict:
-    rag: AppState = app.state.rag
+@app.post("/v1/ask", response_model=AskResponse, dependencies=[Depends(require_api_access)])
+def ask_v1(req: AskRequest, request: Request) -> AskResponse:
+    return _ask(req, request)
+
+
+@app.post(
+    "/ask",
+    response_model=AskResponse,
+    dependencies=[Depends(require_api_access)],
+    deprecated=True,
+)
+def ask_compat(req: AskRequest, request: Request) -> AskResponse:
+    return _ask(req, request)
+
+
+def _health(request: Request) -> HealthResponse:
+    rag: AppState = request.app.state.rag
     cfg = rag.ensure()
-    out: dict = {"service": "pv-rag-agent", "config": {
-        "ollama_url": cfg.ollama_url,
-        "answer_model": cfg.answer_model,
-        "embed_model": cfg.embed_model,
-    }}
     try:
-        retriever = rag.get_retriever()
-        out["index"] = retriever.stats()
-    except RuntimeError as e:
-        out["index"] = {"error": str(e)}
-    client = rag.get_client()
-    out["ollama_up"] = client.is_up()
-    if out["ollama_up"]:
-        try:
-            out["ollama_models"] = client.list_models()
-        except Exception:
-            out["ollama_models"] = None
-    return out
-
-
-@app.post("/reindex")
-def reindex() -> dict:
-    rag: AppState = app.state.rag
-    cfg = rag.ensure()
-    with rag.lock:
-        stats = build_index(cfg, client=rag.get_client())
-        rag.reset_retriever()
-    result = stats.as_dict()
-    result["warning"] = (
-        "Index ohne Dense-Vektoren aufgebaut (Ollama-Embedding nicht verfügbar) — "
-        "BM25-only. 'ollama pull " + cfg.embed_model + "' prüfen und erneut reindexen."
-        if stats.embed_error else None
+        index = rag.get_retriever().stats()
+        index_ok = True
+    except RuntimeError:
+        index = {"available": False}
+        index_ok = False
+    ollama_up = rag.get_client().is_up()
+    return HealthResponse(
+        status="ok" if index_ok and ollama_up else "degraded",
+        index=index,
+        ollama_up=ollama_up,
+        authentication_enabled=bool(cfg.api_key),
     )
-    return result
+
+
+@app.get("/v1/health", response_model=HealthResponse)
+def health_v1(request: Request) -> HealthResponse:
+    return _health(request)
+
+
+@app.get("/health", response_model=HealthResponse, deprecated=True)
+def health_compat(request: Request) -> HealthResponse:
+    return _health(request)
+
+
+def _reindex(request: Request) -> dict:
+    rag: AppState = request.app.state.rag
+    cfg = rag.ensure()
+    with rag.maintenance_lock:
+        try:
+            stats = build_index(cfg, client=rag.get_client())
+            rag.reset_retriever()
+        except Exception as exc:
+            request_id = _request_id(request)
+            logger.exception("Reindex fehlgeschlagen request_id=%s", request_id)
+            raise HTTPException(
+                status_code=503,
+                detail="Der Index konnte nicht neu aufgebaut werden.",
+            ) from exc
+    result = stats.as_dict()
+    result["api_version"] = API_VERSION
+    result["request_id"] = _request_id(request)
+    result["warning"] = (
+        "Index ohne Dense-Vektoren aufgebaut; Embedding-Dienst prüfen und erneut reindexen."
+        if stats.embed_error
+        else None
+    )
+    return result
+
+
+@app.post("/v1/reindex", dependencies=[Depends(require_admin_access)])
+def reindex_v1(request: Request) -> dict:
+    return _reindex(request)
+
+
+@app.post(
+    "/reindex",
+    dependencies=[Depends(require_admin_access)],
+    deprecated=True,
+)
+def reindex_compat(request: Request) -> dict:
+    return _reindex(request)

pv-agent/agent/cli.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import argparse
+import ipaddress
 import json
 import sys
 
@@ -80,7 +81,23 @@ def _cmd_eval(args: argparse.Namespace, cfg: Config) -> int:
     return run_eval(cfg, args)
 
 
+def _is_loopback_bind(host: str) -> bool:
+    if host.casefold() == "localhost":
+        return True
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
 def _cmd_serve(args: argparse.Namespace, cfg: Config) -> int:
+    if not _is_loopback_bind(args.host) and not cfg.api_key:
+        print(
+            "[Fehler] Nicht-lokaler API-Bind ohne PV_API_KEY abgelehnt. "
+            "Service-Key setzen oder nur an 127.0.0.1 binden.",
+            file=sys.stderr,
+        )
+        return 2
     import uvicorn
     uvicorn.run("agent.api:app", host=args.host, port=cfg.port, log_level="info")
     return 0

pv-agent/agent/config.py

@@ -89,9 +89,11 @@ class Config:
 
     # Service
     port: int = 8080
+    api_key: str = ""          # leer = nur fuer lokale Entwicklung ohne Auth
+    admin_api_key: str = ""    # leer = api_key auch fuer /reindex verwenden
 
     @classmethod
-    def from_env(cls) -> "Config":
+    def from_env(cls) -> Config:
         d = cls()
         return cls(
             kb_dir=_env_str("PV_KB_DIR", d.kb_dir),
@@ -122,4 +124,6 @@ class Config:
             survey_blocks=_env_int("PV_SURVEY_BLOCKS", d.survey_blocks),
             temporal_boost=_env_float("PV_TEMPORAL_BOOST", d.temporal_boost),
             port=_env_int("PV_PORT", d.port),
+            api_key=_env_str("PV_API_KEY", d.api_key),
+            admin_api_key=_env_str("PV_ADMIN_API_KEY", d.admin_api_key),
         )

pv-agent/agent/generate.py

@@ -296,8 +296,14 @@ def answer_question(
             "regenerations": regenerations,
             "latency_ms": round((time.perf_counter() - t0) * 1000),
             "draft": draft,
+            "answer_type": qtype,
+            "planned": planned,
             "planned_queries": [
-                {"text": sq.text, "stand_year": sq.stand_year}
+                {
+                    "text": sq.text,
+                    "stand_year": sq.stand_year,
+                    "scope": sq.scope,
+                }
                 for sq in sub_queries
             ],
         }

pv-agent/agent/retrieve.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import json
 import sqlite3
+import threading
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -66,11 +67,16 @@ class Retriever:
             raise RuntimeError(
                 f"index fehlt ({self.db_path}) — zuerst 'python -m agent.cli ingest' ausführen"
             )
-        self._con = sqlite3.connect(self.db_path)
+        # FastAPI fuehrt synchrone Endpunkte in Worker-Threads aus. Die
+        # Verbindung wird deshalb thread-uebergreifend verwendet; ein RLock
+        # serialisiert die kurzen SQLite-/Matrix-Abschnitte je Retriever.
+        self._lock = threading.RLock()
+        self._con = sqlite3.connect(self.db_path, check_same_thread=False)
         self._con.row_factory = sqlite3.Row
         self._client = client
         self._owns_client = client is None
         self._mat: np.ndarray | None = None
+        self._mat_norm: np.ndarray | None = None
         self._mat_chunk_ids: list[int] | None = None
         self._scope_cache: dict[str, set[int]] = {}
         row = self._con.execute("SELECT MIN(stand), MAX(stand) FROM chunks").fetchone()
@@ -78,31 +84,33 @@ class Retriever:
         self._stand_max = int((row[1] or "2026-01").replace("-", ""))
 
     def close(self) -> None:
-        self._con.close()
-        if self._owns_client and self._client is not None:
-            self._client.close()
+        with self._lock:
+            self._con.close()
+            if self._owns_client and self._client is not None:
+                self._client.close()
 
     # -- Index-Kennzahlen ---------------------------------------------------
 
     def stats(self) -> dict:
-        n_chunks = self._con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
-        n_entries = self._con.execute(
-            "SELECT COUNT(DISTINCT entry_id) FROM chunks"
-        ).fetchone()[0]
-        n_vec = self._con.execute(
-            "SELECT COUNT(*) FROM vectors WHERE model = ?",
-            (self.cfg.embed_model,),
-        ).fetchone()[0]
-        meta = dict(self._con.execute("SELECT key, value FROM meta").fetchall())
-        return {
-            "n_entries": n_entries,
-            "n_chunks": n_chunks,
-            "n_vectors": n_vec,
-            "dense_available": n_vec > 0 and not self.cfg.embed_off,
-            "stand_min": str(self._stand_min),
-            "stand_max": str(self._stand_max),
-            "built_at": meta.get("built_at"),
-        }
+        with self._lock:
+            n_chunks = self._con.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
+            n_entries = self._con.execute(
+                "SELECT COUNT(DISTINCT entry_id) FROM chunks"
+            ).fetchone()[0]
+            n_vec = self._con.execute(
+                "SELECT COUNT(*) FROM vectors WHERE model = ?",
+                (self.cfg.embed_model,),
+            ).fetchone()[0]
+            meta = dict(self._con.execute("SELECT key, value FROM meta").fetchall())
+            return {
+                "n_entries": n_entries,
+                "n_chunks": n_chunks,
+                "n_vectors": n_vec,
+                "dense_available": n_vec > 0 and not self.cfg.embed_off,
+                "stand_min": str(self._stand_min),
+                "stand_max": str(self._stand_max),
+                "built_at": meta.get("built_at"),
+            }
 
     # -- Einzelverfahren ----------------------------------------------------
 
@@ -141,7 +149,12 @@ class Retriever:
         if self.cfg.embed_off:
             return {}
         self._ensure_matrix()
-        if self._mat is None or len(self._mat) == 0:
+        if (
+            self._mat is None
+            or self._mat_norm is None
+            or self._mat_chunk_ids is None
+            or len(self._mat) == 0
+        ):
             return {}
         if self._client is None:
             self._client = OllamaClient(
@@ -166,7 +179,7 @@ class Retriever:
         for i in order:
             if len(out) >= limit:
                 break
-            cid = self._mat_chunk_ids[i]
+            cid = self._mat_chunk_ids[int(i)]
             if scope_ids is not None and cid not in scope_ids:
                 continue
             out[cid] = float(sims[i])
@@ -392,16 +405,17 @@ class Retriever:
         deterministisch in Direktzahlung und Alternativen zerlegt. Das gilt
         auch für den Offline-Retrieval-Eval, der keinen LLM-Planer aufruft.
         """
-        n = n_entries or self.cfg.context_blocks
-        latest_year = str(self._stand_max)[:4]
-        deterministic = decision_support_plan(question, default_year=latest_year)
-        if deterministic:
-            sub_queries, _qtype = deterministic
-            return self.search_multi(sub_queries, n_entries=n)
-        fused, bm_all, dn_all = self._fuse_queries(
-            [SubQuery(text=question)], self.cfg.candidate_pool
-        )
-        return self._context_from_fused(fused, n, bm_all, dn_all)
+        with self._lock:
+            n = n_entries or self.cfg.context_blocks
+            latest_year = str(self._stand_max)[:4]
+            deterministic = decision_support_plan(question, default_year=latest_year)
+            if deterministic:
+                sub_queries, _qtype = deterministic
+                return self.search_multi(sub_queries, n_entries=n)
+            fused, bm_all, dn_all = self._fuse_queries(
+                [SubQuery(text=question)], self.cfg.candidate_pool
+            )
+            return self._context_from_fused(fused, n, bm_all, dn_all)
 
     def search_multi(
         self, sub_queries: list, n_entries: int | None = None
@@ -412,44 +426,45 @@ class Retriever:
         seiner besten Quelle hinein — sonst dominieren Einträge, die in
         mehreren Sub-Queries mittelgut matchen). Temporal-Intent: kv-
         Einträge im gefragten Geltungsjahr erhalten temporal_boost."""
-        n = n_entries or self.cfg.context_blocks
-        if (
-            len(sub_queries) == 1
-            and sub_queries[0].scope is None
-            and sub_queries[0].stand_year is None
-        ):
-            return self.search(sub_queries[0].text, n_entries=n)
-        pool = self.cfg.candidate_pool
-        fused_total: dict[int, float] = {}
-        bm_all: set[int] = set()
-        dn_all: set[int] = set()
-        per_query_entries: list[list[str]] = []
-        for sq in sub_queries:
-            bm, dn, fused_q = self._search_one(sq, pool)
-            bm_all.update(bm)
-            dn_all.update(dn)
-            for cid, score in fused_q.items():
-                fused_total[cid] = fused_total.get(cid, 0.0) + score
-            rows_q = self._fetch_chunks(list(fused_q))
-            entry_best: dict[str, float] = {}
-            for cid, score in fused_q.items():
-                row = rows_q.get(cid)
-                if row is not None:
-                    eid = row["entry_id"]
-                    entry_best[eid] = max(entry_best.get(eid, 0.0), score)
-            per_query_entries.append(
-                [e for e, _ in sorted(entry_best.items(), key=lambda kv: -kv[1])]
+        with self._lock:
+            n = n_entries or self.cfg.context_blocks
+            if (
+                len(sub_queries) == 1
+                and sub_queries[0].scope is None
+                and sub_queries[0].stand_year is None
+            ):
+                return self.search(sub_queries[0].text, n_entries=n)
+            pool = self.cfg.candidate_pool
+            fused_total: dict[int, float] = {}
+            bm_all: set[int] = set()
+            dn_all: set[int] = set()
+            per_query_entries: list[list[str]] = []
+            for sq in sub_queries:
+                bm, dn, fused_q = self._search_one(sq, pool)
+                bm_all.update(bm)
+                dn_all.update(dn)
+                for cid, score in fused_q.items():
+                    fused_total[cid] = fused_total.get(cid, 0.0) + score
+                rows_q = self._fetch_chunks(list(fused_q))
+                entry_best: dict[str, float] = {}
+                for cid, score in fused_q.items():
+                    row = rows_q.get(cid)
+                    if row is not None:
+                        eid = row["entry_id"]
+                        entry_best[eid] = max(entry_best.get(eid, 0.0), score)
+                per_query_entries.append(
+                    [e for e, _ in sorted(entry_best.items(), key=lambda kv: -kv[1])]
+                )
+            reserved: list[str] = []
+            for entries in per_query_entries:
+                taken = 0
+                for eid in entries:
+                    if taken >= self.cfg.per_query_slots:
+                        break
+                    if eid not in reserved:
+                        reserved.append(eid)
+                        taken += 1
+            years = {sq.stand_year for sq in sub_queries if sq.stand_year}
+            return self._context_from_fused(
+                fused_total, n, bm_all, dn_all, stand_years=years, reserved=reserved
             )
-        reserved: list[str] = []
-        for entries in per_query_entries:
-            taken = 0
-            for eid in entries:
-                if taken >= self.cfg.per_query_slots:
-                    break
-                if eid not in reserved:
-                    reserved.append(eid)
-                    taken += 1
-        years = {sq.stand_year for sq in sub_queries if sq.stand_year}
-        return self._context_from_fused(
-            fused_total, n, bm_all, dn_all, stand_years=years, reserved=reserved
-        )