feat: Barge-in/Turn-Manager und VAD-Aeusserungserkennung (#4 Ausbau)

- Barge-in: Antwort-Turn als abbrechbarer asyncio.Task; {"type":"interrupt"} oder
  neue Eingabe bricht laufende Antwort ab -> interrupted-Event (/ws/chat + /ws/voice)
- VAD (app/audio/vad.py): energie-basierte Stille-Erkennung (reines Python, int16-PCM)
- /ws/voice opt-in {"type":"start","vad":true}: automatisches Aeusserungsende ohne end
- Tests: 52 gruen (+5: VAD-Unit, Barge-in, VAD-Auto-Segmentierung)
- Doku aktualisiert (README, BEDIENUNGSANLEITUNG, Architektur)

Offen (schwere Deps/Dienste): echte partielle Live-Transkripte (Streaming-STT),
WebRTC (aiortc). STT laeuft heute pro Aeusserung.

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

BEDIENUNGSANLEITUNG.md

@@ -280,7 +280,12 @@ Satz fertig ist.
 
 **Sprach-Eingang** (`/ws/voice`): Mikrofon-Audio als binäre Frames senden, dann
 `{"type":"end"}`. Der Server schickt ein `transcript`-Event und danach die Antwort
-wie bei `/ws/chat` (`stream`/`audio_stream` im `end`-Frame möglich).
+wie bei `/ws/chat` (`stream`/`audio_stream` im `end`-Frame möglich). Mit
+`{"type":"start","vad":true,"format":"pcm","sample_rate":16000}` erkennt der Server
+das Äußerungsende automatisch an einer Sprechpause (kein `end` nötig).
+
+**Unterbrechen (Barge-in):** Während der Assistent antwortet, `{"type":"interrupt"}`
+senden — die laufende Antwort wird abgebrochen (`interrupted`-Event).
 
 > **Für lokale Entwicklung** ist in der mitgelieferten `.env` `AUTH_ENABLED=false`
 > gesetzt — dann ist kein Token nötig (anonymer Nutzer).

Docs/voice-assistant-architecture.md

@@ -189,7 +189,8 @@ Device Router (strikt, Singleton); Output-Lifecycle; **Authentifizierung
 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)**; automatisierte Tests.
+über WebSocket (`/ws/voice`: Audio rein → STT → Antwort-Pipeline) mit VAD-Aeusserungs-
+erkennung und Barge-in (`interrupt`)**; automatisierte Tests.
 
 **Platzhalter (Gerüst):** Audio-Endpunkte (`local-default`, `bluetooth`,
 `mobile-ws`, `mobile-webrtc`) liefern leere Chunks — nur Auswahl/Lifecycle sind
@@ -204,7 +205,7 @@ Reihenfolge der Weiterentwicklung:
 1. **(erledigt)** Konfig- & Routing-Fundament: Profile, Device Router, Registry, Pro-Request-Override.
 2. **(erledigt)** Cloud-Fundament: Bearer-Token-Auth, Mehrbenutzer, persistenter SQLite-Store, Mandanten-Trennung, dauerhafte Nutzer-Präferenzen. Offen: Skalierung auf gemeinsamen Store (Postgres/Redis) für mehrere Instanzen.
 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. **(teilweise erledigt)** Echtzeit: WebSocket-Streaming-Chat (`/ws/chat`), **Token-Level-LLM-Streaming (SSE, `stream:true`)**, **Audio-Streaming (chunked TTS satzweise, `audio_stream:true`)** und **Audio-Eingang (`/ws/voice`: binaere Chunks -> STT -> Antwort-Pipeline)** sind umgesetzt. Offen: **partielle Live-Transkripte (Streaming-STT mit VAD)**, **Barge-in/Turn-Manager**, **WebRTC**.
+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. **Resilienz:** Fallback-Policy (remote KI fällt aus → lokaler/alternativer Provider), Metriken/Tracing.
 6. **Betrieb:** Kosten-/Quota-Kontrolle pro Nutzer; Notfall-/Eskalationskonzept (Senioren-Kontext).
 7. **TransportRouter** als eigene lokal/remote-Achse aktivieren.
@@ -227,7 +228,7 @@ voice-assistant-scaffold/
 │   ├── schemas.py         # Pydantic-Modelle
 │   ├── api/               # health, chat, speak, transcribe, devices, sessions, config, admin, me, ws
 │   ├── core/              # orchestrator
-│   ├── audio/             # router, transport_router, endpoints/input|output/*
+│   ├── audio/             # router, transport_router, vad, endpoints/input|output/*
 │   ├── pipeline/          # input_cleaner, spoken_response_adapter, tts_normalizer, sentence_chunker
 │   └── providers/         # stt/ llm/ tts/  (openrouter + lokale Stubs)
 ├── config/                # voice-assistant.example.toml (+ lokale .toml, gitignored)

README.md

@@ -149,8 +149,18 @@ transkribiert (STT), sendet ein `transcript`-Event und durchläuft dann dieselbe
 Antwort-Pipeline wie `/ws/chat` (inkl. `stream`/`audio_stream`). Damit ist
 Sprach-zu-Sprach-Konversation über einen Kanal möglich.
 
-> Partielle Live-Transkripte (Streaming-STT mit VAD), Barge-in/Turn-Manager und
+**VAD (automatische Äußerungserkennung):** Mit `{"type":"start","vad":true,
-> WebRTC sind als nächste Increments vorgesehen (siehe Architektur-Dokument).
+"sample_rate":16000,"format":"pcm"}` segmentiert der Server Äußerungen selbst anhand
+von Stille (energie-basiert, reines Python) — ohne explizites `end`. Optional:
+`vad_silence_ms`, `vad_threshold`.
+
+**Barge-in:** Eine laufende Antwort lässt sich mit `{"type":"interrupt"}` (oder durch
+eine neue Eingabe) abbrechen — der Server stoppt das Streaming und meldet
+`{"type":"interrupted"}`. Wichtig für natürliche Gespräche.
+
+> Echte **partielle Live-Transkripte** (Streaming-STT-Dienst, wortweise während des
+> Sprechens) und **WebRTC** sind als nächste Increments vorgesehen (siehe
+> Architektur-Dokument). Heute läuft STT pro Äußerung.
 
 ## Authentifizierung
 

app/api/ws.py

@@ -3,14 +3,18 @@
 - /ws/chat : Text rein (JSON pro Turn), Antwort als Event-Folge zurueck.
 - /ws/voice: Audio rein (binaere Chunks + Control), Transkription -> selbe Pipeline.
 
-Event-Folge der Antwort: ack -> [token*] -> [audio*] -> semantic -> done.
+Antwort-Events: ack -> [token*] -> [audio*] -> semantic -> done.
 Mit {"stream":true} kommen LLM-Token live, mit {"audio_stream":true} das Audio
 satzweise (chunked TTS). /ws/voice sendet zuvor ein transcript-Event.
 
-Spaeter (eigene Increments): partielle Live-Transkripte (Streaming-STT mit VAD),
+Barge-in: Ein {"type":"interrupt"}-Frame oder eine neue Eingabe bricht eine laufende
-Barge-in/Turn-Manager und WebRTC.
+Antwort ab (-> interrupted-Event). Der Antwort-Turn laeuft als abbrechbarer Task.
+
+Spaeter (eigene Increments): echte partielle Live-Transkripte (Streaming-STT-Dienst),
+WebRTC.
 """
 
+import asyncio
 import json
 
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
@@ -24,6 +28,7 @@ from app.dependencies import (
     resolve_output_endpoint,
 )
 from app.store import SessionOwnershipError
+from app.audio.vad import EnergyVAD
 
 router = APIRouter()
 
@@ -126,6 +131,51 @@ async def _run_turn(websocket, store, user, session_id, route, orchestrator, out
     await websocket.send_json({"type": "done", "audio_format": "pcm", "sample_rate": 24000})
 
 
+async def _cancel_active(task, websocket) -> None:
+    """Bricht einen laufenden Antwort-Turn ab (Barge-in) und meldet 'interrupted'."""
+    if task is None or task.done():
+        return
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+    await websocket.send_json({"type": "interrupted"})
+
+
+async def _chat_turn(websocket, store, user, session_id, text, options):
+    try:
+        route, orchestrator, output = await _resolve(user, session_id, options)
+    except SessionOwnershipError as exc:
+        await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)})
+        return
+    except RoutingError as exc:
+        await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)})
+        return
+    await _run_turn(websocket, store, user, session_id, route, orchestrator, output, text, options)
+
+
+async def _voice_turn(websocket, store, user, session_id, audio, fmt, options):
+    try:
+        route, orchestrator, output = await _resolve(user, session_id, options)
+    except SessionOwnershipError as exc:
+        await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)})
+        return
+    except RoutingError as exc:
+        await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)})
+        return
+    try:
+        transcript = await orchestrator.stt.transcribe(audio, fmt=fmt, language=route.language)
+    except Exception as exc:
+        await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)})
+        return
+    await websocket.send_json({"type": "transcript", "text": transcript})
+    if not transcript or not transcript.strip():
+        await websocket.send_json({"type": "error", "detail": "empty transcript"})
+        return
+    await _run_turn(websocket, store, user, session_id, route, orchestrator, output, transcript, options)
+
+
 @router.websocket("/ws/chat")
 async def ws_chat(websocket: WebSocket, session_id: str | None = None, token: str | None = None):
     user = _authenticate(token)
@@ -134,24 +184,26 @@ async def ws_chat(websocket: WebSocket, session_id: str | None = None, token: st
         return
     await websocket.accept()
     store = get_store()
+    active = None
 
     try:
         while True:
             msg = await websocket.receive_json()
+            if msg.get("type") == "interrupt":
+                await _cancel_active(active, websocket)
+                active = None
+                continue
             text = (msg.get("text") or "").strip()
             if not text:
                 await websocket.send_json({"type": "error", "detail": "empty text"})
                 continue
-            try:
+            await _cancel_active(active, websocket)  # Barge-in bei neuer Eingabe
-                route, orchestrator, output = await _resolve(user, session_id, msg)
+            active = asyncio.create_task(
-            except SessionOwnershipError as exc:
+                _chat_turn(websocket, store, user, session_id, text, msg)
-                await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)})
+            )
-                continue
-            except RoutingError as exc:
-                await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)})
-                continue
-            await _run_turn(websocket, store, user, session_id, route, orchestrator, output, text, msg)
     except WebSocketDisconnect:
+        if active and not active.done():
+            active.cancel()
         return
 
 
@@ -166,15 +218,33 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s
 
     audio_buffer = bytearray()
     fmt = "wav"
+    active = None
+    vad = None
+    vad_options: dict = {}
+
+    async def _start_voice(audio: bytes, options: dict):
+        nonlocal active
+        await _cancel_active(active, websocket)  # Barge-in bei neuer Aeusserung
+        active = asyncio.create_task(
+            _voice_turn(websocket, store, user, session_id, audio, fmt, options)
+        )
 
     try:
         while True:
             message = await websocket.receive()
             if message["type"] == "websocket.disconnect":
+                if active and not active.done():
+                    active.cancel()
                 return
 
             if message.get("bytes") is not None:
                 audio_buffer.extend(message["bytes"])
+                # VAD: Aeusserungsende automatisch erkennen (opt-in via start-Frame).
+                if vad is not None and vad.feed(message["bytes"]):
+                    audio = bytes(audio_buffer)
+                    audio_buffer.clear()
+                    vad.reset()
+                    await _start_voice(audio, vad_options)
                 continue
 
             raw = message.get("text")
@@ -187,9 +257,22 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s
                 continue
 
             ctype = control.get("type")
+            if ctype == "interrupt":
+                await _cancel_active(active, websocket)
+                active = None
+                continue
             if ctype == "start":
                 audio_buffer.clear()
                 fmt = control.get("format", "wav")
+                if control.get("vad"):
+                    vad = EnergyVAD(
+                        sample_rate=control.get("sample_rate", 16000),
+                        threshold=control.get("vad_threshold", 500.0),
+                        silence_ms=control.get("vad_silence_ms", 700.0),
+                    )
+                    vad_options = control
+                else:
+                    vad = None
                 continue
             if ctype != "end":
                 continue
@@ -198,35 +281,12 @@ async def ws_voice(websocket: WebSocket, session_id: str | None = None, token: s
                 await websocket.send_json({"type": "error", "detail": "no audio received"})
                 continue
 
-            try:
+            audio = bytes(audio_buffer)
-                route, orchestrator, output = await _resolve(user, session_id, control)
+            audio_buffer.clear()
-            except SessionOwnershipError as exc:
+            if vad is not None:
-                await websocket.send_json({"type": "error", "status": 403, "detail": str(exc)})
+                vad.reset()
-                audio_buffer.clear()
+            await _start_voice(audio, control)
-                continue
-            except RoutingError as exc:
-                await websocket.send_json({"type": "error", "status": 422, "detail": str(exc)})
-                audio_buffer.clear()
-                continue
-
-            try:
-                transcript = await orchestrator.stt.transcribe(
-                    bytes(audio_buffer), fmt=fmt, language=route.language
-                )
-            except Exception as exc:
-                await websocket.send_json({"type": "error", "status": 502, "detail": str(exc)})
-                audio_buffer.clear()
-                continue
-            finally:
-                audio_buffer.clear()
-
-            await websocket.send_json({"type": "transcript", "text": transcript})
-            if not transcript or not transcript.strip():
-                await websocket.send_json({"type": "error", "detail": "empty transcript"})
-                continue
-
-            await _run_turn(
-                websocket, store, user, session_id, route, orchestrator, output, transcript, control
-            )
     except WebSocketDisconnect:
+        if active and not active.done():
+            active.cancel()
         return

app/audio/vad.py

@@ -0,0 +1,59 @@
+"""Einfache energie-basierte Sprachaktivitaetserkennung (VAD).
+
+Reines Python (stdlib `array`), arbeitet auf s16le-PCM (mono). Erkennt das Ende
+einer Aeusserung anhand andauernder Stille nach erkannter Sprache. Damit kann der
+Server in /ws/voice Aeusserungen automatisch segmentieren, ohne dass der Client
+ein explizites Ende-Signal schickt.
+
+Hinweis: Das ersetzt keinen echten Streaming-STT-Dienst (keine wortweisen
+Teil-Transkripte) - es bestimmt nur die Aeusserungsgrenzen.
+"""
+
+import array
+import math
+
+
+def rms(pcm: bytes) -> float:
+    """Lautstaerke (RMS) eines s16le-PCM-Puffers; 0.0 bei leerem Puffer."""
+    usable = len(pcm) - (len(pcm) % 2)
+    if usable <= 0:
+        return 0.0
+    samples = array.array("h")
+    samples.frombytes(pcm[:usable])
+    if not samples:
+        return 0.0
+    return math.sqrt(sum(s * s for s in samples) / len(samples))
+
+
+class EnergyVAD:
+    def __init__(self, sample_rate: int = 16000, threshold: float = 500.0, silence_ms: float = 700.0):
+        self.sample_rate = sample_rate
+        self.threshold = threshold
+        self.silence_ms = silence_ms
+        self._speech_started = False
+        self._silence_ms = 0.0
+
+    def feed(self, pcm: bytes) -> bool:
+        """Verarbeitet einen Audio-Chunk.
+
+        Liefert True, sobald nach erkannter Sprache genug Stille (silence_ms)
+        vergangen ist - die Aeusserung gilt dann als beendet.
+        """
+        level = rms(pcm)
+        n_samples = len(pcm) // 2
+        chunk_ms = (n_samples / self.sample_rate) * 1000.0 if self.sample_rate else 0.0
+
+        if level >= self.threshold:
+            self._speech_started = True
+            self._silence_ms = 0.0
+            return False
+
+        if self._speech_started:
+            self._silence_ms += chunk_ms
+            if self._silence_ms >= self.silence_ms:
+                return True
+        return False
+
+    def reset(self) -> None:
+        self._speech_started = False
+        self._silence_ms = 0.0

tests/test_realtime.py

@@ -0,0 +1,129 @@
+import array
+import asyncio
+
+from fastapi.testclient import TestClient
+
+import app.dependencies as deps
+from app.main import app
+from app.audio.vad import rms, EnergyVAD
+
+client = TestClient(app)
+
+
+def _pcm(amplitude: int, n_samples: int) -> bytes:
+    return array.array("h", [amplitude] * n_samples).tobytes()
+
+
+# --- VAD-Unit-Tests --------------------------------------------------------
+
+def test_rms_silence_vs_loud():
+    assert rms(b"") == 0.0
+    assert rms(_pcm(0, 100)) == 0.0
+    assert rms(_pcm(3000, 100)) > 2000
+
+
+def test_energy_vad_ends_after_speech_then_silence():
+    vad = EnergyVAD(sample_rate=16000, threshold=500, silence_ms=300)
+    loud = _pcm(3000, 1600)   # 100 ms Sprache
+    silent = _pcm(0, 1600)    # 100 ms Stille
+    assert vad.feed(loud) is False
+    assert vad.feed(silent) is False   # 100 ms
+    assert vad.feed(silent) is False   # 200 ms
+    assert vad.feed(silent) is True    # 300 ms -> Ende
+
+
+def test_energy_vad_ignores_silence_without_speech():
+    vad = EnergyVAD(sample_rate=16000, threshold=500, silence_ms=100)
+    for _ in range(10):
+        assert vad.feed(_pcm(0, 1600)) is False
+
+
+# --- Barge-in --------------------------------------------------------------
+
+def _install_slow_stream(monkeypatch):
+    class SlowLLM:
+        async def complete(self, text, history=None, session_id=None):
+            return "fertig"
+
+        async def stream(self, text, history=None, session_id=None):
+            for i in range(200):
+                await asyncio.sleep(0.005)
+                yield f"t{i} "
+
+    class StubTTS:
+        async def synthesize(self, text, voice=None, audio_format="pcm"):
+            return b"A"
+
+    monkeypatch.setitem(deps.LLM_REGISTRY, "slow", lambda s: SlowLLM())
+    monkeypatch.setitem(deps.TTS_REGISTRY, "stub", lambda s: StubTTS())
+    return {
+        "llm_provider": "slow",
+        "tts_provider": "stub",
+        "output_endpoint": "loopback",
+        "stream": True,
+    }
+
+
+def test_ws_chat_interrupt_cancels_response(monkeypatch):
+    base = _install_slow_stream(monkeypatch)
+    with client.websocket_connect("/ws/chat") as ws:
+        ws.send_json({"text": "Hallo", **base})
+        assert ws.receive_json()["type"] == "ack"
+        assert ws.receive_json()["type"] == "token"  # Antwort laeuft
+
+        ws.send_json({"type": "interrupt"})
+
+        event = None
+        for _ in range(500):
+            event = ws.receive_json()
+            if event["type"] in ("interrupted", "done"):
+                break
+        assert event["type"] == "interrupted"  # abgebrochen, nicht fertig
+
+
+# --- VAD im WebSocket ------------------------------------------------------
+
+def _install_voice_stubs(monkeypatch):
+    class STT:
+        async def transcribe(self, audio_bytes, fmt, language=None):
+            return "ok"
+
+    class LLM:
+        async def complete(self, text, history=None, session_id=None):
+            return "antwort"
+
+    class TTS:
+        async def synthesize(self, text, voice=None, audio_format="pcm"):
+            return b"A"
+
+    monkeypatch.setitem(deps.STT_REGISTRY, "s", lambda x: STT())
+    monkeypatch.setitem(deps.LLM_REGISTRY, "l", lambda x: LLM())
+    monkeypatch.setitem(deps.TTS_REGISTRY, "t", lambda x: TTS())
+    return {"stt_provider": "s", "llm_provider": "l", "tts_provider": "t", "output_endpoint": "loopback"}
+
+
+def test_ws_voice_vad_auto_segments_utterance(monkeypatch):
+    opts = _install_voice_stubs(monkeypatch)
+    start = {
+        "type": "start",
+        "vad": True,
+        "sample_rate": 16000,
+        "vad_silence_ms": 200,
+        "format": "pcm",
+        **opts,
+    }
+    loud = _pcm(3000, 1600)    # 100 ms Sprache
+    silent = _pcm(0, 1600)     # je 100 ms Stille
+
+    with client.websocket_connect("/ws/voice") as ws:
+        ws.send_json(start)
+        ws.send_bytes(loud)
+        ws.send_bytes(silent)   # 100 ms
+        ws.send_bytes(silent)   # 200 ms -> VAD-Ende, Turn startet automatisch
+
+        transcript = ws.receive_json()
+        assert transcript["type"] == "transcript" and transcript["text"] == "ok"
+        assert ws.receive_json()["type"] == "ack"
+        assert ws.receive_json()["type"] == "semantic"
+        assert ws.receive_bytes() == b"A"
+        assert ws.receive_json()["type"] == "done"