feat: Tageskontingent und Notfall-Eskalation (#6)

- Quota (app/quota.py): Anfragen pro Nutzer/Tag (usage-Tabelle), DAILY_REQUEST_LIMIT,
  pro Nutzer via prefs uebersteuerbar; 429 (REST) bzw. error-Event (WS); Metrik
- Notfall (app/safety/emergency.py): heuristische Erkennung (de/en); Log im Store
  (emergency_events) + optionaler Webhook (best-effort) + X-Emergency/emergency-Event;
  Metrik emergency_total; Notfaelle umgehen das Kontingent
- verdrahtet in chat/speak/transcribe + WS-Turns
- Tests: 64 gruen (+6); Doku aktualisiert (README, BEDIENUNGSANLEITUNG, Architektur, .env.example)

Hinweis: Notfall-Erkennung ist eine Heuristik (kein Lebensretter); erkannte Texte
sind sensibel -> DSGVO beachten.

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

.env.example

@@ -42,3 +42,7 @@ LOCAL_LLM_MODEL=llama3.1
 # STT_FALLBACK=faster-whisper
 # LLM_FALLBACK=local-openai-compatible
 # TTS_FALLBACK=piper
+
+# --- Betrieb: Kontingent & Notfall -----------------------------------------
+DAILY_REQUEST_LIMIT=0          # Anfragen pro Nutzer/Tag (0 = unbegrenzt)
+# EMERGENCY_WEBHOOK_URL=https://example.org/alert  # optionale Eskalation

BEDIENUNGSANLEITUNG.md

@@ -313,6 +313,27 @@ curl http://localhost:8080/api/metrics?format=prometheus
 Enthält Request-Zahlen/-Laufzeiten, Pipeline-Stufen (`stt`/`llm`/`tts`) und
 Fallback-/Fehlerzähler. Die Werte gelten pro laufendem Prozess.
 
+**Tageskontingent** (Kostenbremse) in `.env`:
+
+```
+DAILY_REQUEST_LIMIT=200    # Anfragen pro Nutzer/Tag; 0 = unbegrenzt
+```
+
+Bei Überschreitung antwortet der Dienst mit `429`. Notfall-Eingaben werden nie
+blockiert.
+
+**Notfall-Eskalation:** Erkennt der Dienst in einer Chat-/Sprach-Eingabe ein
+Notlagen-Signal (z. B. „Schmerzen in der Brust", „gestürzt", „kann nicht atmen"),
+protokolliert er das, macht es sichtbar (`X-Emergency` / `emergency`-Event) und ruft
+optional einen Webhook auf:
+
+```
+# EMERGENCY_WEBHOOK_URL=https://example.org/alert
+```
+
+> ⚠️ Nur eine **Heuristik** — kein Ersatz für einen echten Notruf. Erkannte Texte
+> sind sensibel; auf Einwilligung und Datenschutz achten.
+
 ---
 
 ## 12. Fehlerbehebung

Docs/voice-assistant-architecture.md

@@ -191,7 +191,9 @@ Store, fließt ins LLM)**; **Langzeit-Erinnerungen pro Nutzer (als LLM-Kontext)*
 **WebSocket-Streaming-Chat (`/ws/chat`) inkl. Token-Level-LLM-Streaming (SSE,
 `stream:true`) und satzweisem Audio-Streaming (chunked TTS, `audio_stream:true`)**; **Sprach-Eingang
 über WebSocket (`/ws/voice`: Audio rein → STT → Antwort-Pipeline) mit VAD-Aeusserungs-
-erkennung und Barge-in (`interrupt`)**; automatisierte Tests.
+erkennung und Barge-in (`interrupt`)**; **Resilienz (Fallback-Ketten je Modul,
+In-Memory-Metriken `/api/metrics`), Tageskontingent pro Nutzer und heuristische
+Notfall-Eskalation**; automatisierte Tests.
 
 **Platzhalter (Gerüst):** Audio-Endpunkte (`local-default`, `bluetooth`,
 `mobile-ws`, `mobile-webrtc`) liefern leere Chunks — nur Auswahl/Lifecycle sind
@@ -208,7 +210,7 @@ Reihenfolge der Weiterentwicklung:
 3. **(erledigt)** Konversationsgedächtnis: Kurzzeit-Gesprächsverlauf pro Session + Langzeit-Erinnerungen pro Nutzer (manuell gepflegt, als LLM-Kontext). Offen: **automatische** Extraktion/Zusammenfassung von Erinnerungen aus Gesprächen.
 4. **(weitgehend erledigt)** Echtzeit: WebSocket-Streaming-Chat (`/ws/chat`), **Token-Level-LLM-Streaming (SSE, `stream:true`)**, **Audio-Streaming (chunked TTS satzweise, `audio_stream:true`)**, **Audio-Eingang (`/ws/voice`)**, **Barge-in/Turn-Manager (`interrupt` bricht laufende Antwort ab)** und **VAD-Aeusserungserkennung (energie-basiert, opt-in)** sind umgesetzt. Offen: **echte partielle Live-Transkripte (Streaming-STT-Dienst, wortweise)** und **WebRTC (aiortc)** — beide brauchen schwere Abhaengigkeiten/Dienste. Heute laeuft STT pro Aeusserung.
 5. **(weitgehend erledigt)** Resilienz: Fallback-Ketten je Modul (`*_FALLBACK`, Provider faellt aus → naechster) und In-Memory-Metriken (`/api/metrics`: Request/Latenz, Pipeline-Stufen, Fallback/Fehler; JSON + Prometheus). Offen: verteiltes Tracing, Alerting.
-6. **Betrieb:** Kosten-/Quota-Kontrolle pro Nutzer; Notfall-/Eskalationskonzept (Senioren-Kontext).
+6. **(weitgehend erledigt)** Betrieb: Tageskontingent pro Nutzer (`DAILY_REQUEST_LIMIT`, 429) und heuristische Notfall-Eskalation (Erkennung -> Log + optionaler Webhook + Flag/Event). Offen: echte Klassifikation statt Schluesselwort-Heuristik, Telefon-/Angehoerigen-Integration, Abrechnung.
 7. **TransportRouter** als eigene lokal/remote-Achse aktivieren.
 
 **Datenschutz (querschnittlich, ab sofort mitdenken):** Senioren-Sprachdaten sind
@@ -226,6 +228,8 @@ voice-assistant-scaffold/
 │   ├── store.py           # Persistenz: Store-Interface + SQLiteStore (Nutzer/Sessions/Verlauf)
 │   ├── auth.py            # Bearer-Token-Auth (require_user) + Admin-Schutz
 │   ├── metrics.py         # In-Memory-Metriken (Counter/Timer, JSON + Prometheus)
+│   ├── quota.py           # Tageskontingent pro Nutzer (Kostenkontrolle)
+│   ├── safety/            # emergency.py: heuristische Notfall-Erkennung/-Eskalation
 │   ├── errors.py          # RoutingError -> HTTP 422
 │   ├── schemas.py         # Pydantic-Modelle
 │   ├── api/               # health, chat, speak, transcribe, devices, sessions, config, admin, me, ws

README.md

@@ -17,6 +17,7 @@ Praktische Bedienung: [`BEDIENUNGSANLEITUNG.md`](BEDIENUNGSANLEITUNG.md).
 - **Routing auf jeder Ebene:** Default → Profil → Nutzer → Session → Request
 - **Authentifizierung** (Bearer-Token) + persistente Nutzer/Sessions (SQLite)
 - **Resilienz:** Fallback-Ketten je Modul (Provider fällt aus → nächster) + Metriken
+- **Betrieb:** Tageskontingent pro Nutzer (`429`) + heuristische Notfall-Eskalation
 - **Gesprächsgedächtnis pro Session:** Verlauf wird gespeichert und fließt ins LLM
 - **Langzeit-Erinnerungen pro Nutzer:** dauerhafte Fakten/Vorlieben als LLM-Kontext
 - **WebSocket-Streaming-Chat** (`/ws/chat`) als Echtzeit-Transport
@@ -188,6 +189,31 @@ curl http://localhost:8080/api/metrics
 curl http://localhost:8080/api/metrics?format=prometheus
 ```
 
+## Kontingent & Notfall-Eskalation
+
+**Tageskontingent** pro Nutzer begrenzt die Kosten (Cloud-LLM/TTS). Bei Überschreitung
+`HTTP 429` (bzw. `error`-Event über WebSocket):
+
+```bash
+DAILY_REQUEST_LIMIT=200        # 0 = unbegrenzt; pro Nutzer/Tag
+```
+Pro Nutzer übersteuerbar via `prefs.daily_request_limit` (siehe `PUT /api/me/prefs`).
+
+**Notfall-Eskalation:** `/api/chat` und `/ws/chat` prüfen die Nutzereingabe heuristisch
+auf Notlagen-Signale (medizinisch, Selbstgefährdung, Hilferuf — de/en). Bei Treffer
+wird der Vorfall protokolliert, optional ein Webhook ausgelöst und das Signal sichtbar
+gemacht (`X-Emergency`-Header / `emergency`-Feld / WebSocket-`emergency`-Event). Eine
+Notfall-Eingabe umgeht das Kontingent (wird nie geblockt).
+
+```bash
+EMERGENCY_WEBHOOK_URL=https://example.org/alert   # optional, Benachrichtigung
+```
+
+> ⚠️ Die Erkennung ist eine **Schlüsselwort-Heuristik** — kein verlässlicher
+> Lebensretter und kein Ersatz für einen echten Notruf. Sie kann Notlagen verpassen
+> oder Fehlalarme auslösen. Erkannte Texte sind hochsensibel (DSGVO: Einwilligung,
+> Aufbewahrung, Zugriff beachten).
+
 ## Authentifizierung
 
 Standardmäßig (`AUTH_ENABLED=true`) sind `chat`/`speak`/`transcribe`/`sessions`/`me`

app/api/chat.py

@@ -13,6 +13,8 @@ from app.dependencies import (
     resolve_output_endpoint,
     get_store,
 )
+from app.quota import enforce_quota, record_usage, QuotaExceededError
+from app.safety.emergency import handle_emergency
 from app.schemas import ChatRequest
 
 router = APIRouter()
@@ -77,6 +79,15 @@ async def chat(
         )
         llm_context = [{"role": "system", "content": memory_text}] + llm_context
 
+    # Notfall-Erkennung zuerst (immer eskalieren, auch bei Quota-Limit).
+    emergency = handle_emergency(user, payload.text, store)
+
+    if emergency is None:
+        try:
+            enforce_quota(user, store)
+        except QuotaExceededError as exc:
+            raise HTTPException(status_code=429, detail=str(exc))
+
     try:
         trace, audio = await orchestrator.chat_text(
             payload.text,
@@ -88,6 +99,8 @@ async def chat(
     except Exception as exc:
         raise HTTPException(status_code=502, detail=str(exc))
 
+    record_usage(user, store, len(payload.text) + len(trace.semantic_response or ""))
+
     # Turn persistieren (User-Eingabe + semantische Antwort) fuer das Gedaechtnis.
     if session_id:
         store.append_message(session_id, user.id, "user", payload.text)
@@ -101,6 +114,7 @@ async def chat(
                 "route": route.as_dict(),
                 "history_len": len(conversation),
                 "memories_len": len(memories),
+                "emergency": emergency,
                 "trace": {
                     "raw_transcript": trace.raw_transcript,
                     "cleaned_transcript": trace.cleaned_transcript,
@@ -119,4 +133,6 @@ async def chat(
         "X-Audio-Sample-Width": "16",
         **_route_headers(route),
     }
+    if emergency:
+        headers["X-Emergency"] = emergency["category"]
     return StreamingResponse(BytesIO(audio), media_type="audio/pcm", headers=headers)

app/api/speak.py

@@ -11,7 +11,9 @@ from app.dependencies import (
     resolve_route,
     build_orchestrator,
     resolve_output_endpoint,
+    get_store,
 )
+from app.quota import enforce_quota, record_usage, QuotaExceededError
 from app.schemas import SpeakRequest
 
 router = APIRouter()
@@ -42,6 +44,12 @@ async def speak(
     except RoutingError as exc:
         raise HTTPException(status_code=422, detail=str(exc))
 
+    store = get_store()
+    try:
+        enforce_quota(user, store)
+    except QuotaExceededError as exc:
+        raise HTTPException(status_code=429, detail=str(exc))
+
     try:
         audio = await orchestrator.speak_only(
             payload.text,
@@ -49,6 +57,7 @@ async def speak(
             language=route.language,
             output=output,
         )
+        record_usage(user, store, len(payload.text))
 
         headers = {
             "Content-Language": route.language,

app/api/transcribe.py

@@ -7,7 +7,9 @@ from app.dependencies import (
     resolve_route,
     build_orchestrator,
     resolve_input_endpoint,
+    get_store,
 )
+from app.quota import enforce_quota, record_usage, QuotaExceededError
 
 router = APIRouter()
 
@@ -39,6 +41,12 @@ async def transcribe(
     except RoutingError as exc:
         raise HTTPException(status_code=422, detail=str(exc))
 
+    store = get_store()
+    try:
+        enforce_quota(user, store)
+    except QuotaExceededError as exc:
+        raise HTTPException(status_code=429, detail=str(exc))
+
     content = await file.read()
     suffix = (file.filename or "audio.wav").rsplit(".", 1)[-1].lower()
 
@@ -52,4 +60,6 @@ async def transcribe(
     except Exception as exc:
         raise HTTPException(status_code=502, detail=str(exc))
 
+    record_usage(user, store, len(trace.raw_transcript or ""))
+
     return {"route": route.as_dict(), "trace": trace.model_dump()}

app/api/ws.py

@@ -29,6 +29,8 @@ from app.dependencies import (
 )
 from app.store import SessionOwnershipError
 from app.audio.vad import EnergyVAD
+from app.quota import enforce_quota, record_usage, QuotaExceededError
+from app.safety.emergency import handle_emergency
 
 router = APIRouter()
 
@@ -75,6 +77,17 @@ async def _run_turn(websocket, store, user, session_id, route, orchestrator, out
         )
         llm_context = [{"role": "system", "content": memory_text}] + llm_context
 
+    # Notfall-Erkennung zuerst (immer eskalieren, auch bei Quota-Limit).
+    emergency = handle_emergency(user, text, store)
+    if emergency:
+        await websocket.send_json({"type": "emergency", "category": emergency["category"]})
+    else:
+        try:
+            enforce_quota(user, store)
+        except QuotaExceededError as exc:
+            await websocket.send_json({"type": "error", "status": 429, "detail": str(exc)})
+            return
+
     await websocket.send_json({"type": "ack", "route": route.as_dict()})
 
     voice = options.get("voice") or settings.openrouter_tts_voice
@@ -119,6 +132,8 @@ async def _run_turn(websocket, store, user, session_id, route, orchestrator, out
         await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)})
         return
 
+    record_usage(user, store, len(text) + len(trace.semantic_response or ""))
+
     if session_id:
         store.append_message(session_id, user.id, "user", text)
         store.append_message(session_id, user.id, "assistant", trace.semantic_response)

app/config.py

@@ -127,6 +127,8 @@ class Settings(BaseSettings):
     stt_fallback: str = ""  # kommaseparierte Provider-Namen (Fallback-Kette)
     llm_fallback: str = ""
     tts_fallback: str = ""
+    daily_request_limit: int = 0  # 0 = unbegrenzt; Anfragen pro Nutzer pro Tag
+    emergency_webhook_url: str = ""  # optionaler Eskalations-Webhook
     model_config = SettingsConfigDict(
         env_file=ENV_FILE, case_sensitive=False, extra="ignore"
     )

app/quota.py

@@ -0,0 +1,40 @@
+"""Pro-Nutzer-Tageskontingent (Kostenkontrolle).
+
+Limit aus Settings (`daily_request_limit`), pro Nutzer ueber `prefs.daily_request_limit`
+ueberschreibbar. 0 bedeutet unbegrenzt.
+"""
+
+from app.config import settings, Settings
+from app.metrics import metrics
+
+
+class QuotaExceededError(Exception):
+    def __init__(self, limit: int, count: int):
+        self.limit = limit
+        self.count = count
+        super().__init__(f"Daily request limit reached ({count}/{limit})")
+
+
+def effective_limit(user, cfg: Settings = settings) -> int:
+    pref = user.prefs.get("daily_request_limit") if user and user.prefs else None
+    if pref is not None:
+        try:
+            return int(pref)
+        except (TypeError, ValueError):
+            pass
+    return cfg.daily_request_limit
+
+
+def enforce_quota(user, store, cfg: Settings = settings) -> None:
+    """Wirft QuotaExceededError, wenn das Tageslimit erreicht ist."""
+    limit = effective_limit(user, cfg)
+    if limit and limit > 0:
+        count = store.get_request_count(user.id)
+        if count >= limit:
+            metrics.inc("quota_exceeded_total")
+            raise QuotaExceededError(limit, count)
+
+
+def record_usage(user, store, units: int = 0) -> None:
+    store.add_usage(user.id, units)
+    metrics.inc("turns_total")

app/safety/emergency.py

@@ -0,0 +1,84 @@
+"""Heuristische Notfall-Erkennung und Eskalation (Senioren-Kontext).
+
+WICHTIG: Schluesselwort-Heuristik, KEIN Ersatz fuer eine echte Klassifikation.
+Sie kann Notlagen verpassen oder Fehlalarme ausloesen. Die erkannten Textauszuege
+sind hochsensibel und werden bewusst protokolliert (DSGVO beachten: Einwilligung,
+Aufbewahrung, Zugriff).
+"""
+
+import asyncio
+
+import httpx
+
+from app.config import settings, Settings
+from app.metrics import metrics
+
+# Phrasen je Kategorie (de/en), bewusst eher spezifisch gegen Fehlalarme.
+_PATTERNS: dict[str, list[str]] = {
+    "medical": [
+        "brustschmerz", "schmerzen in der brust", "kann nicht atmen", "keine luft",
+        "atemnot", "herzinfarkt", "schlaganfall", "bewusstlos", "gestuerzt", "gestürzt",
+        "gefallen und komme nicht hoch", "starke blutung",
+        "chest pain", "can't breathe", "cannot breathe", "heart attack", "stroke",
+        "i fell and can't", "bleeding badly",
+    ],
+    "self_harm": [
+        "nicht mehr leben", "mich umbringen", "selbstmord", "suizid", "will sterben",
+        "kill myself", "end my life", "suicide", "want to die",
+    ],
+    "help": [
+        "notruf", "notarzt", "krankenwagen", "ruf einen arzt", "es brennt",
+        "call an ambulance", "call 911", "call 112",
+    ],
+}
+
+
+def detect(text: str):
+    """Liefert (category, matched_phrase) oder None."""
+    if not text:
+        return None
+    low = text.lower()
+    for category, phrases in _PATTERNS.items():
+        for phrase in phrases:
+            if phrase in low:
+                return category, phrase
+    return None
+
+
+async def _fire_webhook(url: str, user, category: str, snippet: str) -> None:
+    try:
+        async with httpx.AsyncClient(timeout=5) as client:
+            await client.post(
+                url,
+                json={
+                    "user_id": user.id,
+                    "display_name": user.display_name,
+                    "category": category,
+                    "text": snippet,
+                },
+            )
+    except Exception:  # noqa: BLE001 - best effort, darf den Chat nicht brechen
+        metrics.inc("emergency_webhook_error_total")
+
+
+def handle_emergency(user, text: str, store, cfg: Settings = settings):
+    """Erkennt, protokolliert und eskaliert ein Notfall-Signal.
+
+    Gibt {"category", "matched"} zurueck, wenn etwas erkannt wurde, sonst None.
+    Der Webhook (falls konfiguriert) wird nicht-blockierend ausgeloest.
+    """
+    match = detect(text)
+    if not match:
+        return None
+    category, phrase = match
+    snippet = text[:500]
+    store.log_emergency(user.id, category, snippet)
+    metrics.inc("emergency_total", {"category": category})
+    if cfg.emergency_webhook_url:
+        try:
+            asyncio.get_running_loop().create_task(
+                _fire_webhook(cfg.emergency_webhook_url, user, category, snippet)
+            )
+        except RuntimeError:
+            pass  # kein laufender Event-Loop (z. B. im Test) -> Webhook ueberspringen
+    return {"category": category, "matched": phrase}

app/store.py

@@ -97,6 +97,18 @@ class Store(ABC):
     def delete_memory(self, user_id: str, memory_id: int) -> bool:
         """Loescht eine Erinnerung des Nutzers. True, wenn etwas geloescht wurde."""
 
+    @abstractmethod
+    def get_request_count(self, user_id: str, day: str | None = None) -> int:
+        """Anzahl der Anfragen des Nutzers am angegebenen Tag (Default: heute, UTC)."""
+
+    @abstractmethod
+    def add_usage(self, user_id: str, units: int = 0, day: str | None = None) -> int:
+        """Zaehlt eine Anfrage (+units) und liefert die neue Tages-Anfragezahl."""
+
+    @abstractmethod
+    def log_emergency(self, user_id: str, category: str, snippet: str) -> None:
+        """Protokolliert ein erkanntes Notfall-Signal (sensibel!)."""
+
 
 class SQLiteStore(Store):
     def __init__(self, db_path: str):
@@ -146,6 +158,20 @@ class SQLiteStore(Store):
                 );
                 CREATE INDEX IF NOT EXISTS idx_memories_user
                     ON memories(user_id, id);
+                CREATE TABLE IF NOT EXISTS usage (
+                    user_id TEXT NOT NULL,
+                    day TEXT NOT NULL,
+                    requests INTEGER NOT NULL DEFAULT 0,
+                    units INTEGER NOT NULL DEFAULT 0,
+                    PRIMARY KEY (user_id, day)
+                );
+                CREATE TABLE IF NOT EXISTS emergency_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id TEXT NOT NULL,
+                    category TEXT NOT NULL,
+                    snippet TEXT NOT NULL,
+                    created_at TEXT NOT NULL
+                );
                 """
             )
 
@@ -301,3 +327,41 @@ class SQLiteStore(Store):
                 (memory_id, user_id),
             )
             return cur.rowcount > 0
+
+    # ----- Nutzung / Quota --------------------------------------------------
+    @staticmethod
+    def _today() -> str:
+        return datetime.now(timezone.utc).date().isoformat()
+
+    def get_request_count(self, user_id: str, day: str | None = None) -> int:
+        day = day or self._today()
+        with self._connect() as conn:
+            row = conn.execute(
+                "SELECT requests FROM usage WHERE user_id = ? AND day = ?",
+                (user_id, day),
+            ).fetchone()
+        return int(row["requests"]) if row else 0
+
+    def add_usage(self, user_id: str, units: int = 0, day: str | None = None) -> int:
+        day = day or self._today()
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT INTO usage (user_id, day, requests, units) VALUES (?, ?, 1, ?)"
+                " ON CONFLICT(user_id, day) DO UPDATE SET"
+                " requests = requests + 1, units = units + excluded.units",
+                (user_id, day, units),
+            )
+            row = conn.execute(
+                "SELECT requests FROM usage WHERE user_id = ? AND day = ?",
+                (user_id, day),
+            ).fetchone()
+        return int(row["requests"])
+
+    # ----- Notfall-Protokoll ------------------------------------------------
+    def log_emergency(self, user_id: str, category: str, snippet: str) -> None:
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT INTO emergency_events (user_id, category, snippet, created_at)"
+                " VALUES (?, ?, ?, ?)",
+                (user_id, category, snippet, _now()),
+            )

tests/test_quota_safety.py

@@ -0,0 +1,81 @@
+from fastapi.testclient import TestClient
+
+import app.dependencies as deps
+from app.main import app
+from app.config import settings
+from app.safety.emergency import detect
+
+client = TestClient(app)
+
+
+def _stub_chat(monkeypatch, answer="ok"):
+    class StubLLM:
+        async def complete(self, text, history=None, session_id=None):
+            return answer
+
+    class StubTTS:
+        async def synthesize(self, text, voice=None, audio_format="pcm"):
+            return b"A"
+
+    monkeypatch.setitem(deps.LLM_REGISTRY, "l", lambda s: StubLLM())
+    monkeypatch.setitem(deps.TTS_REGISTRY, "t", lambda s: StubTTS())
+    return {"llm_provider": "l", "tts_provider": "t"}
+
+
+# --- Notfall-Erkennung (Einheit) ------------------------------------------
+
+def test_detect_categories():
+    assert detect("Ich habe Brustschmerzen")[0] == "medical"
+    assert detect("Bitte ruf einen Arzt")[0] == "help"
+    assert detect("I want to die")[0] == "self_harm"
+    assert detect("Wie wird das Wetter morgen?") is None
+    assert detect("") is None
+
+
+# --- Quota -----------------------------------------------------------------
+
+def test_quota_blocks_after_limit(monkeypatch):
+    base = _stub_chat(monkeypatch)
+    monkeypatch.setattr(settings, "daily_request_limit", 1)
+    body = {"text": "Hallo", **base}
+    assert client.post("/api/chat?debug=true", json=body).status_code == 200
+    assert client.post("/api/chat?debug=true", json=body).status_code == 429
+
+
+def test_quota_unlimited_by_default(monkeypatch):
+    base = _stub_chat(monkeypatch)
+    body = {"text": "Hallo", **base}
+    for _ in range(3):
+        assert client.post("/api/chat?debug=true", json=body).status_code == 200
+
+
+# --- Notfall ueber Endpunkt ------------------------------------------------
+
+def test_emergency_surfaced_in_response(monkeypatch):
+    base = _stub_chat(monkeypatch, answer="Bleiben Sie ruhig.")
+    resp = client.post(
+        "/api/chat?debug=true",
+        json={"text": "Ich habe starke Schmerzen in der Brust", **base},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["emergency"]["category"] == "medical"
+
+
+def test_emergency_bypasses_quota(monkeypatch):
+    base = _stub_chat(monkeypatch)
+    monkeypatch.setattr(settings, "daily_request_limit", 1)
+    body = {**base}
+    assert client.post("/api/chat?debug=true", json={"text": "hallo", **body}).status_code == 200
+    assert client.post("/api/chat?debug=true", json={"text": "hallo", **body}).status_code == 429
+    rescue = client.post("/api/chat?debug=true", json={"text": "ich kann nicht atmen", **body})
+    assert rescue.status_code == 200
+    assert rescue.json()["emergency"]["category"] == "medical"
+
+
+def test_ws_emergency_event(monkeypatch):
+    base = _stub_chat(monkeypatch, answer="Ruhig bleiben, Hilfe kommt.")
+    with client.websocket_connect("/ws/chat") as ws:
+        ws.send_json({"text": "Ich bin gestürzt und komme nicht hoch", **base})
+        event = ws.receive_json()
+        assert event["type"] == "emergency" and event["category"] == "medical"
+        assert ws.receive_json()["type"] == "ack"