AuthorPascal DateJuly 14, 2026 Rev1.0

Robot Voice Chat — Brain Architecture Reference

Reference / archived

This is the detailed, source-of-truth architecture reference for the brain (robot-voice-chat). It began as a from-scratch rewrite specification; the rewrite has since landed, so it now describes how the brain is actually built. For the conventions distilled from it, see Coding guidelines.

---

Table of Contents

1. Overview
2. System Overview & Functional Requirements
3. Technology Stack
4. Project Structure
5. Backend Architecture
6. Frontend Architecture
7. Configuration Management
8. Service Layer Specifications
9. API Specification
10. WebSocket Protocol
11. Data Models
12. Security Architecture
13. Observability & Logging
14. Testing Strategy
15. Deployment Architecture
16. Non-Functional Requirements
17. Key Behavioral Contracts

---

1. Overview

The brain is a multilingual, voice- and text-controlled robot chat application. It is built on production-grade frameworks with strict separation of concerns, type safety, security hardening, structured logging, and a modern frontend.

Target Platform: NVIDIA Jetson Orin NX (ARM64 / aarch64), Ubuntu 22.04, headless operation. Must also run on macOS for local development.

Core product behavior: - Push-to-talk or Voice Activity Detection (VAD) triggers audio capture - Speech transcribed via Groq or OpenAI Whisper - LLM (Groq) processes transcript with tool-calling capability - LLM either selects a pre-recorded answer or generates a new response - Response synthesized via OpenAI TTS or local Piper TTS - Robot actions (gestures) executed in parallel with speech playback via MCP protocol - Real-time status updates via WebSocket to browser-based control interface - Support for multiple conversation scenarios switchable at runtime

---

2. System Overview & Functional Requirements

2.1 Input Modes

Push-to-Talk (PTT) Mode (keyboard/presenter clicker): - Listen for a configurable key event (e.g., KEY_PAGEDOWN, space) - Start recording on press, stop on release - On Linux headless: use evdev library, scan /dev/input/event* - On macOS/display systems: fall back to pynput - Critical bug to preserve: ignore key-release events from clickers that fire both PRESS+RELEASE on a single physical press. Implement toggle-on-press-only mode.

Voice Activity Detection (VAD) Mode: - Continuously monitor audio via pw-record (PipeWire) subprocess - Energy-based detection using RMS on 16-bit PCM samples - Pre-speech ring buffer: 0.5s of audio kept before detection fires (prevents clipping) - State machine: LISTENING → RECORDING → COOLDOWN - Configurable: energy_threshold (RMS amplitude), silence_duration (seconds of silence to end utterance), min_speech_duration (reject short noise bursts) - Robot prevents self-triggering: while in SPEAKING or PROCESSING state, discard VAD events

2.2 Audio Pipeline

[INPUT: keyboard/VAD] → [RECORD: PyAudio or arecord] → [WAV bytes]
→ [STT: Groq Whisper or OpenAI Whisper] → {text, language}
→ [LLM: Groq llama with tool calling] → {text_response, tool_calls[]}
→ [TOOL DISPATCH] → parallel: [TTS + playback] + [MCP gesture]

Recording: - Primary: PyAudio with named device lookup (case-insensitive substring match on device name) - Fallback: arecord -D plughw:{card},0 subprocess (full path /usr/bin/arecord for systemd) - Sample rate: 16 kHz for STT recording, 48 kHz for VAD monitoring - Output: raw WAV bytes in memory (no temp files for recording)

Playback (fallback chain, try in order): 1. mpg123 -a plughw:{card},0 {file} (MP3, ALSA explicit card) 2. ffplay -nodisp -autoexit {file} 3. aplay -D plughw:{card},0 {file} (WAV only) 4. paplay {file} (PulseAudio) 5. afplay {file} (macOS) 6. PyAudio streaming fallback

2.3 Language Support

• Languages: English (en), German (de), French (fr), Italian (it)
• Whisper returns full language names: "english", "german", "french"
• Must map to ISO 639-1 codes: en, de, fr, it
• Language used for: TTS voice selection, predefined answer file selection
• Config option: force a fixed language (e.g., language: "en") or null for auto-detect

2.4 LLM Tool Calling

Predefined Answers (high-priority, lowest latency): - LLM receives tool definition: use_predefined_answer(answer_id: enum) - answer_id enum values auto-generated from predefined_answers.yaml for the active scenario - If LLM calls this tool: play pre-recorded MP3 from audio/predefined/{answer_id}_{lang}.mp3 - Fallback: {answer_id}_en.mp3 if language variant missing

Non-Blocking MCP Tools (gesture tools, fire-and-forget): - Execute in ThreadPoolExecutor alongside TTS playback - Example: g1_robot.execute_face_wave, g1_robot.execute_high_five - Tool result not fed back to LLM (fire-and-forget) - Errors logged but do not interrupt speech

Blocking MCP Tools (sensor/perception tools, require LLM follow-up): - Defined in config as blocking_tools list (e.g., vlm_mcp.get_scene_description) - Sequence: 1. Clear pending TTS audio queue 2. Synthesize short acknowledgment phrase ("Let me check…") 3. Execute tool call in parallel with acknowledgment playback 4. Wait for result (timeout: 30s) 5. Make follow-up LLM call with tool result as context 6. Synthesize and play new response

2.5 Streaming Response Synthesis

• LLM response streamed token by token
• Split into synthesis chunks at sentence boundaries (., !, ?, \n)
• TTS synthesis triggered per chunk as stream arrives
• Audio files queued and consumed by background playback thread
• Result: first audio plays while LLM/TTS still generating later sentences

2.6 Predefined Answers Configuration

YAML-defined per scenario:

scenarios:
  "Belimed":
    - id: "introduction"
      description: "If asked to introduce yourself"
      text_en: "Hello everyone, I'm Milo..."
      text_de: "..."   # optional
      text_fr: "..."   # optional

Audio files must be generated separately (batch TTS script) and stored as: audio/predefined/{answer_id}_{lang}.mp3

2.7 MCP Protocol Integration

• Connects to MCP servers over stdio (subprocess-based)
• Auto-discovers tools from each server via list_tools() RPC
• Tool names prefixed: {server_name}.{tool_name} (dot notation)
• All async operations run in dedicated background event loop thread
• Timeout per tool call: 30 seconds
• response_text parameter auto-injected into all tool schemas for parallel execution context

2.8 Web Interface

Control interface accessible from a remote browser (e.g., tablet, laptop on same network): - Real-time status display (idle / recording / processing / speaking) - Live transcript and response display - Scenario switcher - Device status indicators (microphone, speaker, keyboard — green/red) - Configurable per-deployment theming (logo, color scheme) - Debug mode: show push-to-talk button, show tool call logs - Service restart button

2.9 Conversation Scenarios

Runtime-switchable personality/context profiles: - Defined in system_prompts.yaml as array of {name, system_prompt} - Each scenario has its own predefined answers - Switch via web UI or API — takes effect on next utterance

---

3. Technology Stack

3.1 Backend

Component	Technology	Version	Rationale
Web framework	FastAPI	≥0.115	Async-native, automatic OpenAPI docs, native WebSocket, Pydantic integration, type-safe
WebSocket	python-socketio with socketio.AsyncServer (ASGI)	≥5.11	Socket.IO protocol; ASGI-native; auto-reconnect on client
ASGI server	Uvicorn with uvloop + httptools	≥0.30	Production ASGI server; uvloop replaces asyncio loop for ~2-4× throughput
Config composition	Hydra (hydra-core) + OmegaConf	≥1.3	Config groups for swappable components, override syntax, env-var resolvers. Drives config-based DI (see §5.6)
Config validation	Pydantic v2 structured configs	≥2.0	Hydra-composed config validated through Pydantic dataclasses at startup — type safety + fail-fast
Dependency injection	Hydra instantiate (_target_) + FastAPI Depends()	—	Config selects concrete implementation; instantiate builds it; Depends() injects into routes (see §5.6)
Data validation	Pydantic v2	≥2.0	All data models, request/response schemas
Logging	structlog	≥24.0	Structured JSON logs, context binding, async support
LLM/STT client	groq SDK (AsyncGroq)	≥0.32	Native async client — no thread-pool wrapping; true non-blocking streaming
OpenAI client	openai SDK (AsyncOpenAI)	≥1.0	Native async client for STT + TTS
JSON	orjson	≥3.10	Fastest JSON (de)serialization; FastAPI ORJSONResponse default
HTTP transport	httpx (shared AsyncClient, HTTP/2, keep-alive pool)	≥0.27	Connection reuse across API calls — eliminates per-request TLS handshake latency
Audio I/O	sounddevice (PortAudio) + PyAudio fallback	latest	sounddevice gives NumPy-native callback streams (lower copy overhead); subprocess ALSA as last resort
MCP client	mcp library	latest	Model Context Protocol client
VAD audio	numpy	≥1.24	Vectorized RMS computation
Rate limiting	slowapi	≥0.1.9	Per-endpoint rate limiting (Starlette/FastAPI)
Testing	pytest + pytest-asyncio + httpx + respx	latest	Async tests + HTTP mocking at the transport layer
Linting/format	ruff	≥0.4	Fast linter + formatter
Type checking	mypy --strict	≥1.10	Static type checking, enforced in CI

Not needed (deliberately excluded): - No ORM/database (stateless application, no persistence required) - No message broker (asyncio task orchestration sufficient for single-node) - No external DI container library (e.g. dependency-injector): Hydra's instantiate already is a config-driven DI container, and FastAPI Depends() covers request-scoped injection. Adding a third system would be redundant.

Why Hydra over plain Pydantic Settings? The original ask is "an abstract interface for each capability, with DI selecting the concrete implementation from config." That is exactly what Hydra config groups + instantiate(_target_=...) provide out of the box: switching stt=groq → stt=openai on the command line or in YAML swaps the entire concrete class with zero code change. Pydantic Settings validates values but cannot select and construct polymorphic implementations. We therefore use both: Hydra for composition + instantiation, Pydantic for validating the composed result. See §7 and §5.6.

3.2 Frontend

Component	Technology	Version	Rationale
Build tool	Vite	≥5.0	Fast HMR, ES modules, ARM64 build support
Framework	React	≥18	Industry standard, large ecosystem
Language	TypeScript	≥5.0	Type safety, better IDE support
Styling	Tailwind CSS	≥3.4	Utility-first, no CSS bundle bloat, consistent design tokens
Components	shadcn/ui	latest	Radix UI primitives + Tailwind; accessible, unstyled, copy-paste pattern
Icons	Lucide React	latest	Consistent icon set (included in shadcn/ui)
WebSocket	socket.io-client	≥4.7	Matches server-side Socket.IO
State management	Zustand	≥4.0	Lightweight, simple, no boilerplate; sufficient for this SPA
HTTP client	TanStack Query (React Query)	≥5.0	Server state management for API calls
Animations	Framer Motion	≥11	Status indicator animations, transitions
Audio viz	Canvas API (native)	—	No library needed for simple bar animation

3.3 Infrastructure

Component	Technology	Rationale
Process supervisor	systemd service	Native Linux init; already used
Container (optional)	Docker + docker-compose	Optional local dev; NOT for Jetson production (overhead)
Package management (Python)	uv	10-100× faster than pip, lockfile support, virtual env management
Package management (Node)	pnpm	Fast, disk-efficient
Secret management	.env file + environment variables	Never in YAML; validated at startup

---

4. Project Structure

robot-voice-chat/
├── backend/
│   ├── pyproject.toml              # Project metadata, deps, tool config (ruff, mypy, pytest)
│   ├── uv.lock                     # Locked dependencies
│   ├── .env.example                # Template for secrets (committed)
│   ├── .env                        # Actual secrets (gitignored)
│   │
│   ├── src/
│   │   └── robot_voice_chat/
│   │       ├── __init__.py
│   │       ├── main.py             # App entrypoint, lifespan, mount routes
│   │       ├── composition.py      # CompositionRoot.build() — assembles object graph
│   │       │
│   │       ├── config/
│   │       │   ├── __init__.py
│   │       │   ├── schema.py       # Pydantic structured-config dataclasses (AppConfig)
│   │       │   └── loader.py       # Hydra compose + Pydantic validate; scenario/answer YAML
│   │       │
│   │       ├── models/
│   │       │   ├── __init__.py
│   │       │   ├── audio.py        # AudioDevice, DeviceStatus, TranscriptionResult
│   │       │   ├── chat.py         # Message, ToolCall, LLMResponse, LLMChunk
│   │       │   ├── scenario.py     # Scenario, PredefinedAnswer
│   │       │   └── events.py       # WebSocket event payloads (all typed)
│   │       │
│   │       ├── services/
│   │       │   ├── __init__.py
│   │       │   ├── audio/
│   │       │   │   ├── base.py             # ABCs: Recorder, AudioPlayer, DeviceProbe
│   │       │   │   ├── pyaudio_recorder.py
│   │       │   │   ├── sounddevice_recorder.py
│   │       │   │   ├── arecord_recorder.py
│   │       │   │   ├── fallback_recorder.py # composite (Chain of Responsibility)
│   │       │   │   ├── subprocess_player.py # mpg123/ffplay/aplay/paplay/afplay chain
│   │       │   │   ├── pyaudio_player.py
│   │       │   │   └── device_probe.py
│   │       │   ├── input/
│   │       │   │   ├── base.py             # ABC: UtteranceSource; KeyTrigger ABC
│   │       │   │   ├── ptt_source.py       # PushToTalkSource (KeyTrigger + Recorder)
│   │       │   │   ├── vad_source.py       # VadSource (energy-based, pw-record)
│   │       │   │   ├── evdev_trigger.py
│   │       │   │   └── pynput_trigger.py
│   │       │   ├── llm/
│   │       │   │   ├── base.py             # ABC: LLMProvider
│   │       │   │   └── groq.py             # GroqLLM (AsyncGroq)
│   │       │   ├── stt/
│   │       │   │   ├── base.py             # ABC: STTProvider
│   │       │   │   ├── groq.py             # GroqSTT
│   │       │   │   └── openai.py           # OpenAISTT
│   │       │   ├── tts/
│   │       │   │   ├── base.py             # ABC: TTSProvider
│   │       │   │   ├── openai.py           # OpenAITTS (AsyncOpenAI)
│   │       │   │   ├── piper.py            # PiperTTS
│   │       │   │   └── espeak.py           # EspeakTTS
│   │       │   └── tools/
│   │       │       ├── base.py             # ABC: ToolBackend; NullToolBackend
│   │       │       └── mcp.py              # McpToolBackend (event-loop thread)
│   │       │
│   │       ├── pipeline/
│   │       │   ├── __init__.py
│   │       │   ├── orchestrator.py     # Core audio pipeline (was process_audio)
│   │       │   ├── tool_dispatcher.py  # Handles all tool execution modes
│   │       │   └── playback_queue.py   # Async audio queue + worker
│   │       │
│   │       ├── api/
│   │       │   ├── __init__.py
│   │       │   ├── router.py           # Aggregate all routers
│   │       │   ├── routes/
│   │       │   │   ├── scenarios.py    # /api/scenarios
│   │       │   │   ├── config.py       # /api/config
│   │       │   │   ├── devices.py      # /api/device_status
│   │       │   │   └── admin.py        # /api/restart, /api/health
│   │       │   └── dependencies.py     # FastAPI Depends() providers
│   │       │
│   │       ├── websocket/
│   │       │   ├── __init__.py
│   │       │   ├── server.py           # Socket.IO server setup
│   │       │   ├── handlers.py         # connect, disconnect, start/stop_recording
│   │       │   └── emitter.py          # EventEmitter service (typed emit calls)
│   │       │
│   │       └── state/
│   │           ├── __init__.py
│   │           └── app_state.py        # AppState dataclass (thread-safe)
│   │
│   └── tests/
│       ├── conftest.py
│       ├── unit/
│       │   ├── test_settings.py
│       │   ├── test_audio_service.py
│       │   ├── test_vad_service.py
│       │   ├── test_tool_dispatcher.py
│       │   └── test_orchestrator.py
│       └── integration/
│           ├── test_api_routes.py
│           └── test_websocket.py
│
├── frontend/
│   ├── package.json
│   ├── pnpm-lock.yaml
│   ├── tsconfig.json
│   ├── vite.config.ts
│   ├── tailwind.config.ts
│   ├── index.html
│   │
│   └── src/
│       ├── main.tsx
│       ├── App.tsx
│       ├── types/
│       │   ├── events.ts           # WebSocket event types (mirrors backend models)
│       │   └── api.ts              # API response types
│       ├── stores/
│       │   ├── appStore.ts         # Zustand: status, scenario, messages
│       │   └── deviceStore.ts      # Zustand: device status
│       ├── hooks/
│       │   ├── useSocket.ts        # Socket.IO connection + event handling
│       │   ├── useRecording.ts     # PTT keyboard handler
│       │   ├── useDeviceStatus.ts  # Polling + React Query
│       │   └── useScenarios.ts     # React Query
│       ├── components/
│       │   ├── ui/                 # shadcn/ui generated components
│       │   ├── StatusIndicator.tsx
│       │   ├── ConversationFeed.tsx
│       │   ├── MessageBubble.tsx
│       │   ├── ScenarioSelector.tsx
│       │   ├── DeviceStatus.tsx
│       │   ├── AudioVisualizer.tsx
│       │   ├── Header.tsx
│       │   └── PushToTalkHint.tsx
│       └── lib/
│           ├── socket.ts           # Singleton Socket.IO client
│           ├── queryClient.ts      # TanStack Query client
│           └── utils.ts            # cn(), text cleaning, etc.
│
├── config/                         # Hydra config tree (config groups = swappable impls)
│   ├── config.yaml                 # Root: defaults list + composition
│   ├── stt/
│   │   ├── groq.yaml               # _target_: ...stt.groq.GroqSTT
│   │   └── openai.yaml             # _target_: ...stt.openai.OpenAISTT
│   ├── llm/
│   │   └── groq.yaml
│   ├── tts/
│   │   ├── openai.yaml
│   │   ├── piper.yaml
│   │   └── espeak.yaml
│   ├── input/
│   │   ├── push_to_talk.yaml       # _target_: ...input.ptt_source.PushToTalkSource
│   │   └── vad.yaml                # _target_: ...input.vad_source.VadSource
│   ├── audio/
│   │   ├── player/
│   │   │   ├── subprocess.yaml
│   │   │   └── pyaudio.yaml
│   │   └── recorder/
│   │       ├── fallback.yaml
│   │       ├── pyaudio.yaml
│   │       └── arecord.yaml
│   ├── tools/
│   │   ├── mcp.yaml                # _target_: ...tools.mcp.McpToolBackend
│   │   └── none.yaml               # _target_: ...tools.base.NullToolBackend
│   ├── deployment/                 # whole-deployment presets (compose everything)
│   │   ├── belimed.yaml            # @package _global_ override bundle
│   │   └── zhaw.yaml
│   ├── system_prompts.yaml         # Conversation scenarios (data, not DI)
│   └── predefined_answers.yaml     # Pre-scripted responses (data, not DI)
│
├── audio/
│   ├── predefined/                 # Pre-recorded MP3s (generated)
│   └── temp/                       # Ephemeral TTS output (gitignored)
│
├── scripts/
│   ├── generate_predefined_audio.py
│   ├── test_audio.py
│   ├── test_keyboard.py
│   └── test_mcp.py
│
├── deploy/
│   ├── robot-voice-chat.service    # systemd unit file template
│   └── sudoers.d                   # sudoers snippet for restart
│
├── .gitignore
├── README.md
└── CLAUDE.md

---

5. Backend Architecture

5.1 Layered Architecture

The backend follows a strict ports-and-adapters (hexagonal) layering. Dependencies point inward only.

┌──────────────────────────────────────────────────────────────┐
│  Transport      api/ (REST routes)   websocket/ (Socket.IO)     │  ← adapters in
├──────────────────────────────────────────────────────────────┤
│  Application    pipeline/  (orchestrator, tool_dispatcher,      │  ← use cases
│                 playback_queue)                                 │
├──────────────────────────────────────────────────────────────┤
│  Domain Ports   services/**/base.py  (ABCs: Recorder,           │  ← interfaces
│                 AudioPlayer, UtteranceSource, STTProvider,       │
│                 LLMProvider, TTSProvider, ToolBackend)           │
├──────────────────────────────────────────────────────────────┤
│  Adapters out   services/**/  (PyAudioRecorder, GroqSTT,        │  ← implementations
│                 OpenAITTS, PiperTTS, McpToolBackend, ...)        │
└──────────────────────────────────────────────────────────────┘

Rule: The application layer depends only on the abstract ports (base.py), never on a concrete adapter. Concrete adapters are chosen at composition time by Hydra (§5.6) and injected. This is what makes "swap STT provider via config, zero code change" possible, and what makes every component independently unit-testable with a fake.

5.2 Abstract Ports (the interfaces)

Every capability that has more than one possible implementation, or that touches hardware/network (and therefore must be fakeable in tests), is defined as an abstract base class. We use abc.ABC (not bare Protocol) for ports we instantiate via Hydra, because _target_ needs a concrete class reference and ABCs give us @abstractmethod enforcement at construction time. Protocol is used only for purely structural, never-instantiated contracts.

# services/audio/base.py
class Recorder(ABC):
    """Low-level capture. THE start/stop-recording abstraction."""
    @abstractmethod
    def start_recording(self) -> None: ...
    @abstractmethod
    def stop_recording(self) -> bytes:        # returns WAV bytes
        ...
    @abstractmethod
    async def aclose(self) -> None: ...

class AudioPlayer(ABC):
    @abstractmethod
    async def play(self, audio: Path | bytes) -> None: ...

class DeviceProbe(ABC):
    """Reports microphone / speaker / keyboard connectivity."""
    @abstractmethod
    async def status(self) -> DeviceStatus: ...

# services/input/base.py
class UtteranceSource(ABC):
    """
    High-level trigger that yields complete user utterances.
    Both push-to-talk (keyboard + Recorder) and VAD implement this.
    The orchestrator depends ONLY on this interface and never knows
    whether input came from a clicker or from voice detection.
    """
    @abstractmethod
    def on_utterance(self, callback: Callable[[bytes], Awaitable[None]]) -> None: ...
    @abstractmethod
    async def start(self) -> None: ...
    @abstractmethod
    async def stop(self) -> None: ...
    @property
    @abstractmethod
    def idle_status(self) -> Literal["idle", "listening"]:
        """'idle' for push-to-talk, 'listening' for VAD — used for status events."""

# services/stt/base.py
class STTProvider(ABC):
    @abstractmethod
    async def transcribe(self, audio: bytes, language: str | None = None) -> TranscriptionResult: ...

# services/llm/base.py
class LLMProvider(ABC):
    @abstractmethod
    def chat_stream(self, messages: list[Message], tools: list[ToolDefinition]) -> AsyncIterator[LLMChunk]: ...
    @abstractmethod
    async def chat(self, messages: list[Message], tools: list[ToolDefinition]) -> LLMResponse: ...

# services/tts/base.py
class TTSProvider(ABC):
    @abstractmethod
    async def synthesize(self, text: str, language: str) -> Path: ...

# services/tools/base.py
class ToolBackend(ABC):
    """Robot-action provider. MCP today; could be HTTP/gRPC tomorrow."""
    @abstractmethod
    def tool_definitions(self) -> list[ToolDefinition]: ...
    @abstractmethod
    async def call(self, tool_name: str, arguments: dict) -> str: ...
    @abstractmethod
    async def aclose(self) -> None: ...

class NullToolBackend(ToolBackend):
    """Used when mcp.enabled = false. No-op, returns empty tool list."""

Concrete adapters per port:

Port	Adapters (selected by config)
Recorder	PyAudioRecorder, SoundDeviceRecorder, ArecordRecorder, FallbackRecorder (composite)
AudioPlayer	SubprocessPlayer (mpg123/ffplay/aplay/paplay/afplay chain), PyAudioPlayer
UtteranceSource	PushToTalkSource (composes KeyTrigger + Recorder), VadSource
STTProvider	GroqSTT, OpenAISTT
LLMProvider	GroqLLM
TTSProvider	OpenAITTS, PiperTTS, EspeakTTS
ToolBackend	McpToolBackend, NullToolBackend

The FallbackRecorder and SubprocessPlayer are themselves composites that hold an ordered list of strategies — the fallback behavior of the PoC becomes an explicit, testable Decorator/Chain instead of inline try/except.

5.3 Config-Driven Dependency Injection

Each adapter declares its dependencies in its constructor — nothing reaches into globals or a service locator. The wiring is produced by Hydra instantiate from the composed config (§7). A single CompositionRoot assembles the object graph once at startup:

# composition.py
@dataclass
class CompositionRoot:
    recorder: Recorder | None         # None in VAD mode
    player: AudioPlayer
    utterance_source: UtteranceSource
    stt: STTProvider
    llm: LLMProvider
    tts: TTSProvider
    tools: ToolBackend
    device_probe: DeviceProbe
    orchestrator: PipelineOrchestrator
    emitter: EventEmitter

async def build(cfg: AppConfig, sio: socketio.AsyncServer) -> CompositionRoot:
    # Hydra instantiate selects + constructs each concrete adapter from `_target_`.
    player   = instantiate(cfg.audio.player)         # e.g. _target_=...SubprocessPlayer
    stt      = instantiate(cfg.stt)                  # _target_ chosen by `stt=groq|openai`
    llm      = instantiate(cfg.llm)
    tts      = instantiate(cfg.tts)
    tools    = await build_tool_backend(cfg.tools)   # async init for MCP
    emitter  = EventEmitter(sio)
    source   = instantiate(cfg.input, recorder=maybe(cfg))   # PTT or VAD
    orch     = PipelineOrchestrator(stt, llm, tts, player, tools, ...)
    return CompositionRoot(...)

instantiate(cfg.stt) reads _target_: robot_voice_chat.services.stt.groq.GroqSTT (or the OpenAI target if stt=openai was composed) and constructs it with the remaining keys as kwargs. Switching implementations is a one-line config change — no factory if/elif ladders anywhere in the codebase.

5.4 Application Entrypoint (main.py)

All initialization happens in FastAPI's lifespan — never at import time. Hydra config is loaded via the Compose API (not the @hydra.main decorator), so the process stays a clean long-running ASGI service with no working-directory hijacking.

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    cfg: AppConfig = load_config()          # Hydra compose + Pydantic validate (§7)
    setup_logging(cfg.logging)
    state = AppState()
    root = await build(cfg, sio)            # CompositionRoot — the object graph

    app.state.root = root
    app.state.app_state = state
    app.state.cfg = cfg

    # Wire the utterance source to the pipeline (works for PTT and VAD identically)
    root.utterance_source.on_utterance(root.orchestrator.process)
    await root.utterance_source.start()

    try:
        yield
    finally:
        await root.utterance_source.stop()
        await root.tools.aclose()
        if root.recorder:
            await root.recorder.aclose()

app = FastAPI(title="Robot Voice Chat", lifespan=lifespan, default_response_class=ORJSONResponse)
socket_app = socketio.ASGIApp(sio, app)     # serve: uvicorn main:socket_app

5.5 Request-Scoped Injection (FastAPI Depends)

The CompositionRoot (singletons) lives in app.state. Routes pull individual ports through thin Depends() providers, so handlers receive interfaces, not concretes:

# api/dependencies.py
def get_root(request: Request) -> CompositionRoot: return request.app.state.root
def get_device_probe(root: CompositionRoot = Depends(get_root)) -> DeviceProbe:
    return root.device_probe

# api/routes/devices.py
@router.get("/api/device_status")
async def device_status(probe: DeviceProbe = Depends(get_device_probe)) -> DeviceStatusResponse:
    return DeviceStatusResponse(data=await probe.status())

This two-tier DI (Hydra builds the graph once; Depends hands out references per request) keeps the hot path allocation-free while preserving full testability — tests override app.dependency_overrides[get_device_probe] with a fake.

5.6 Thread Safety

The audio pipeline runs in a background thread (triggered by keyboard/VAD events). It communicates with the async event loop via: - asyncio.run_coroutine_threadsafe() to emit WebSocket events - threading.Event for synchronization - queue.Queue for audio playback queue - Atomic flags (threading.Event or threading.Lock-protected booleans) for state

AppState must protect all mutable fields with appropriate primitives:

@dataclass
class AppState:
    _lock: threading.Lock = field(default_factory=threading.Lock)
    _current_scenario: str = ""
    is_recording: threading.Event = field(default_factory=threading.Event)
    is_processing: threading.Event = field(default_factory=threading.Event)
    is_speaking: threading.Event = field(default_factory=threading.Event)
    audio_queue: queue.Queue[str | None] = field(default_factory=queue.Queue)

    @property
    def current_scenario(self) -> str:
        with self._lock:
            return self._current_scenario

    @current_scenario.setter
    def current_scenario(self, value: str) -> None:
        with self._lock:
            self._current_scenario = value

5.7 Pipeline Orchestrator

The PipelineOrchestrator replaces the monolithic process_audio() function. It owns the full STT → LLM → TTS → playback lifecycle and depends only on abstract ports:

class PipelineOrchestrator:
    def __init__(
        self,
        stt: STTProvider,
        llm: LLMProvider,
        tts: TTSProvider,
        player: AudioPlayer,
        tools: ToolBackend,
        tool_dispatcher: ToolDispatcher,
        playback: PlaybackQueue,
        emitter: EventEmitter,
        state: AppState,
        cfg: PipelineConfig,
    ): ...

    async def process(self, audio_data: bytes) -> None:
        """
        Full async pipeline — invoked directly as the UtteranceSource callback.
        Runs on the event loop (no run_coroutine_threadsafe needed because
        UtteranceSource adapters already bridge their own threads/subprocesses
        onto the loop via loop.call_soon_threadsafe before invoking this).
        """
        async with self._run_guard():        # sets is_processing, request-id context
            transcript = await self.stt.transcribe(audio_data, self.cfg.language)
            ...                               # stream LLM → chunked TTS → playback

Because the orchestrator only sees STTProvider, LLMProvider, TTSProvider, AudioPlayer, ToolBackend, it is exercised in unit tests with in-memory fakes and zero hardware or network.

Note (see §5.8): As of the Interaction Modalities refactor, the orchestrator no longer takes stt/tts/player/playback directly. Those are encapsulated behind two new ports — InputChannel (input modality) and SpeechOutput (output modality) — and the orchestrator's entrypoint is process_turn(turn: UserTurn), not process(audio_data). The contract above describes the pre-refactor shape; §5.8 supersedes it.

---

5.8 Interaction Modalities (Input Channels & Speech Output)

The PoC and the §5.7 orchestrator hard-code two modality assumptions: input is always microphone audio (always run STT) and output is always speech (always run TTS + playback). To support running as a voice assistant, a text-chat assistant, or both at once — selected purely by config — we lift those two assumptions behind two new ports. The orchestrator core becomes modality-agnostic: text in → text-to-UI (always) + audio-out (optional).

5.8.1 The two new ports

# services/io/base.py

@dataclass(frozen=True)
class UserTurn:
    """A normalized user message, regardless of how it was acquired."""
    text: str
    language: str            # ISO 639-1 (e.g. "en"); voice fills from STT, chat from client

TurnCallback = Callable[[UserTurn], Coroutine[Any, Any, None]]

class InputChannel(ABC):
    """
    Acquires complete user turns and forwards them to the pipeline.
    The orchestrator depends ONLY on this — it never knows whether a turn
    came from a microphone (+STT) or from a typed chat message.
    """
    @abstractmethod
    def on_turn(self, callback: TurnCallback) -> None: ...
    @abstractmethod
    async def start(self) -> None: ...
    @abstractmethod
    async def stop(self) -> None: ...
    @property
    @abstractmethod
    def idle_status(self) -> Literal["idle", "listening"]:
        """'listening' if any sub-channel is VAD (always-on), else 'idle'."""

class SpeechOutput(ABC):
    """
    Renders assistant text as audio. The text itself always reaches the UI via
    the EventEmitter independently of this port; SpeechOutput governs ONLY
    whether (and how) that text is vocalized.
    """
    @abstractmethod
    async def open(self) -> None:
        """Begin a response turn (e.g. start the playback consumer)."""
    @abstractmethod
    async def speak(self, text: str, language: str) -> None:
        """Vocalize a sentence/chunk of assistant text."""
    @abstractmethod
    async def play_audio(self, audio: bytes) -> None:
        """Play raw pre-rendered audio (predefined-answer MP3)."""
    @abstractmethod
    async def interrupt(self) -> None:
        """Drop queued/in-progress speech (barge-in, blocking-tool ack)."""
    @abstractmethod
    async def close(self) -> None:
        """Drain and tear down the response turn."""
    @property
    @abstractmethod
    def produces_audio(self) -> bool:
        """False for text-only output — lets callers skip filesystem/TTS work."""

5.8.2 Adapters

Port	Adapter	Composes / behavior
InputChannel	VoiceInputChannel	Wraps the existing UtteranceSource (§8.2) + STTProvider. On each utterance: transcribe → emit_transcription → fire on_turn. Owns stt_latency_ms and the empty-transcript error. idle_status delegates to the wrapped source.
InputChannel	ChatInputChannel	No hardware. Exposes async submit(text, language) called by the send_message WS handler; emits the user text via emit_transcription (so chat turns render identically to voice turns) then fires on_turn. idle_status = "idle".
InputChannel	CompositeInputChannel	Holds an ordered list of channels; fans out on_turn/start/stop. idle_status = "listening" if any child is, else "idle".
SpeechOutput	VoiceSpeechOutput	Owns PlaybackQueue + TTSProvider. speak = synthesize→enqueue (+ TTFA timing); play_audio = enqueue bytes; interrupt = clear queue; open/close = start/stop the playback consumer. produces_audio = True.
SpeechOutput	TextSpeechOutput	All audio methods are no-ops; produces_audio = False. The UI still shows everything via the emitter.

The pre-existing UtteranceSource (PTT/VAD) is unchanged — it is now an implementation detail of VoiceInputChannel rather than a direct dependency of the orchestrator.

5.8.3 Modality-agnostic core

class PipelineOrchestrator:
    def __init__(self, llm, tools, output: SpeechOutput, dispatcher,
                 registry, emitter, state, idle_status): ...

    async def process_turn(self, turn: UserTurn) -> None:
        if self._state.is_processing.is_set():        # one turn at a time
            return
        # build messages from turn.text; emit status "processing"
        # await output.open()  → stream LLM → emit chunks + output.speak(...)
        # dispatch tools; await output.close(); emit idle_status

ToolDispatcher likewise depends on SpeechOutput instead of tts+playback. For predefined answers it checks output.produces_audio: voice plays the {answer_id}_{lang}.mp3; text emits the localized PredefinedAnswer.text_for_language(lang) as a normal response. The blocking-tool acknowledgement uses output.interrupt() + output.speak(ack) (a no-op under text output).

5.8.4 Config & DI

A typed, fail-fast section selects which modalities are live:

# config/config.yaml
interaction:
  voice: true     # microphone + STT input, and TTS + speaker output
  chat: false     # typed-text input over the WebSocket

# config/schema.py
@dataclass
class InteractionConfig:
    voice: bool = True
    chat: bool = False
    def __post_init__(self) -> None:
        if not (self.voice or self.chat):
            raise ValueError("interaction: at least one of voice/chat must be true")
    @property
    def speech_enabled(self) -> bool:
        return self.voice          # speech output is on iff voice is on

The composition root (§5.3) reads the validated flags and assembles the graph — this is the dependency injection; the orchestrator only ever sees the InputChannel/SpeechOutput ports:

# composition.py (sketch)
output = VoiceSpeechOutput(playback, tts) if interaction.speech_enabled else TextSpeechOutput()

channels: list[InputChannel] = []
if interaction.voice:
    src = instantiate(cfg.input, recorder=recorder)         # PTT or VAD
    channels.append(VoiceInputChannel(src, stt, emitter, state))
if interaction.chat:
    chat_input = ChatInputChannel(emitter)
    channels.append(chat_input)
input_channel = channels[0] if len(channels) == 1 else CompositeInputChannel(channels)
input_channel.on_turn(orchestrator.process_turn)

Voice-only adapters (recorder, stt, tts, player) are constructed only when their modality is active, so a chat-only deployment needs no audio hardware and no STT/TTS API keys. The corresponding CompositionRoot fields become … | None.

---

6. Frontend Architecture

6.1 Build Setup

Vite config serves the built frontend via FastAPI's StaticFiles mount at / in production. In development, Vite dev server proxies /api and WebSocket to the backend.

// vite.config.ts
export default defineConfig({
  plugins: [react()],
  server: {
    proxy: {
      '/api': 'http://localhost:8000',
      '/socket.io': { target: 'http://localhost:8000', ws: true },
    },
  },
  build: { outDir: '../backend/src/robot_voice_chat/static' },
})

FastAPI serves the built frontend:

app.mount("/", StaticFiles(directory="static", html=True), name="static")

6.2 State Architecture

Three isolated concerns:

appStore (Zustand):

interface AppState {
  status: 'idle' | 'recording' | 'processing' | 'speaking' | 'listening'
  currentScenario: string
  messages: Message[]
  isConnected: boolean
  // actions
  setStatus: (status: AppStatus) => void
  addMessage: (message: Message) => void
  setScenario: (scenario: string) => void
}

deviceStore (Zustand):

interface DeviceState {
  microphone: DeviceInfo
  speaker: DeviceInfo
  keyboard: DeviceInfo
  lastChecked: Date | null
}

Server state (TanStack Query): - useScenarios(): GET /api/scenarios, 5 min stale time - useConfig(): GET /api/config, once on mount - useDeviceStatus(): GET /api/device_status, 10s refetch interval

6.3 WebSocket Integration

// lib/socket.ts
const socket = io({ path: '/socket.io', autoConnect: true, reconnection: true })
export default socket

// hooks/useSocket.ts
export function useSocket() {
  const { setStatus, addMessage } = useAppStore()

  useEffect(() => {
    socket.on('status', ({ status }) => setStatus(status))
    socket.on('transcription', ({ text, language }) =>
      addMessage({ type: 'transcription', text, language, timestamp: Date.now() }))
    socket.on('response', ({ text }) =>
      addMessage({ type: 'response', text, timestamp: Date.now() }))
    socket.on('response_chunk', ({ text }) => /* streaming append */)
    socket.on('error', ({ message }) =>
      addMessage({ type: 'error', text: message, timestamp: Date.now() }))
    socket.on('tool_start', ({ tool, args }) => /* optional tool log */)
    socket.on('predefined_answer', ({ answer_id }) => /* optional log */)

    return () => { socket.removeAllListeners() }
  }, [])
}

6.4 Push-to-Talk Handler

// hooks/useRecording.ts
export function useRecording() {
  const { status } = useAppStore()

  useEffect(() => {
    const handleKeyDown = (e: KeyboardEvent) => {
      if (e.code !== 'Space' || e.repeat) return
      if (e.target instanceof HTMLInputElement) return
      if (status === 'idle' || status === 'listening') {
        socket.emit('start_recording')
      }
    }
    const handleKeyUp = (e: KeyboardEvent) => {
      if (e.code !== 'Space') return
      if (status === 'recording') {
        socket.emit('stop_recording')
      }
    }
    window.addEventListener('keydown', handleKeyDown)
    window.addEventListener('keyup', handleKeyUp)
    return () => {
      window.removeEventListener('keydown', handleKeyDown)
      window.removeEventListener('keyup', handleKeyUp)
    }
  }, [status])
}

6.5 Theming

Use Tailwind CSS variables + shadcn/ui theming. Themes defined in tailwind.config.ts as semantic color tokens. Per-deployment theme injected from API config:

// App.tsx
const { data: config } = useConfig()
useEffect(() => {
  if (config?.theme) {
    document.documentElement.setAttribute('data-theme', config.theme)
  }
}, [config?.theme])

CSS variables per theme:

:root { --brand-primary: 265 89% 62%; }       /* default: purple */
[data-theme="belimed"] { --brand-primary: 210 90% 45%; }  /* blue */

6.6 Chat Composer (Interaction Modalities, §5.8)

GET /api/config exposes two booleans, voice_enabled and chat_enabled, mirroring the backend interaction config. The UI adapts:

• chat_enabled → render a ChatComposer (textarea + send button) at the bottom of the conversation feed. On submit it calls socket.emit('send_message', { text, language }) and clears the field. Enter sends; Shift+Enter inserts a newline. The input is disabled while status === 'processing'.
• voice_enabled → render the PushToTalkHint and keep the Space-bar handler active. When voice is disabled, the hint is hidden and Space behaves normally inside the composer.

Typed turns echo back over the existing transcription event, so a chat message renders in the same right-aligned user bubble as a spoken one — the feed is modality-uniform. No new client event types are required for rendering.

---

7. Configuration Management (Hydra + Pydantic)

Configuration does two jobs in this system, and we split them deliberately:

1. Select & construct implementations (which STT? which input mode? which player chain?) → Hydra config groups + instantiate. This is the config-driven DI mechanism (§5.3).
2. Validate & type the resulting values → Pydantic v2 structured configs. Hydra composes raw YAML into an OmegaConf object; we then validate it through Pydantic dataclasses so a bad value fails at startup with a precise error, never at request time.

Critical rule: No secrets in any YAML or in version control. API keys are injected via OmegaConf env resolvers that read the process environment (populated from .env), and are typed SecretStr.

7.1 Config Groups = Swappable Implementations

Each port from §5.2 maps to a Hydra config group (a directory under config/). Each file in that directory is one selectable implementation, and crucially carries the _target_ that instantiate will construct.

# config/stt/groq.yaml
_target_: robot_voice_chat.services.stt.groq.GroqSTT
api_key: ${oc.env:GROQ_API_KEY}      # env resolver — secret never on disk
model: whisper-large-v3-turbo
language: null                        # null = auto-detect

# config/stt/openai.yaml
_target_: robot_voice_chat.services.stt.openai.OpenAISTT
api_key: ${oc.env:OPENAI_API_KEY}
model: whisper-1
language: null

# config/tts/openai.yaml
_target_: robot_voice_chat.services.tts.openai.OpenAITTS
api_key: ${oc.env:OPENAI_API_KEY}
model: gpt-4o-mini-tts
voice: marin
temp_dir: ${paths.temp}

# config/input/vad.yaml
_target_: robot_voice_chat.services.input.vad_source.VadSource
energy_threshold: 300
silence_duration: 1.5
pre_speech_buffer: 0.5
min_speech_duration: 0.3
sample_rate: 48000

# config/input/push_to_talk.yaml
_target_: robot_voice_chat.services.input.ptt_source.PushToTalkSource
trigger_key: KEY_RIGHT
# recorder is injected by the composition root via instantiate(..., recorder=<built>)

# config/audio/recorder/fallback.yaml
_target_: robot_voice_chat.services.audio.fallback_recorder.FallbackRecorder
input_device: "USB Composite Device"
sample_rate: 16000
channels: 1
strategies:                           # ordered Chain of Responsibility
  - _target_: robot_voice_chat.services.audio.pyaudio_recorder.PyAudioRecorder
  - _target_: robot_voice_chat.services.audio.arecord_recorder.ArecordRecorder

# config/tools/mcp.yaml
_target_: robot_voice_chat.services.tools.mcp.McpToolBackend
blocking_tools: ["vlm_mcp.get_scene_description"]
call_timeout_s: 30
servers:
  - name: g1_robot
    command: /home/user/venv/bin/python
    args: ["/home/user/mcp_server/robot.py"]
    env:
      ROS_MASTER_URI: "http://localhost:11311"

7.2 Root Config (config/config.yaml)

The root selects one option per group via the defaults list, then carries non-DI app settings. Swapping any line swaps an entire implementation.

# config/config.yaml
defaults:
  - stt: groq                  # ← change to `openai` to swap STT, no code change
  - llm: groq
  - tts: openai
  - input: push_to_talk        # ← change to `vad` to swap input mode
  - audio/recorder: fallback
  - audio/player: subprocess
  - tools: mcp                 # ← change to `none` to disable robot actions
  - _self_

paths:
  predefined: audio/predefined
  temp: audio/temp

server:
  host: "0.0.0.0"
  port: 8000
  allowed_origins: ["*"]       # tighten in deployment presets

ui:
  theme: default
  logo_url: null
  show_tool_output: false

logging:
  level: INFO
  json_format: true

scenarios_file: config/system_prompts.yaml
predefined_answers_file: config/predefined_answers.yaml

7.3 Deployment Presets (compose everything for one site)

A deployment preset composes a full configuration for a site and lives in config/deployment/. It overrides the defaults list and any values, so a deploy is a single switch.

# config/deployment/belimed.yaml
# @package _global_
defaults:
  - override /input: push_to_talk
  - override /stt: openai
  - override /tools: mcp

audio:
  recorder:
    input_device: "AB13X USB Audio"
ui:
  theme: belimed
  logo_url: "https://example.com/belimed-logo.png"
  show_tool_output: true
server:
  allowed_origins: ["http://192.168.1.50:8000"]

Run it: uvicorn ... with RVC_DEPLOYMENT=belimed, or in dev python -m robot_voice_chat +deployment=belimed. Hydra's CLI override syntax also allows ad-hoc changes without editing files: ... stt=openai tts=piper logging.level=DEBUG.

7.4 Structured Config Schema + Validation

We register Pydantic-typed dataclasses so Hydra knows the shape, and we validate the composed result before building anything. This is the fail-fast gate.

# config/schema.py
from pydantic import SecretStr, Field
from pydantic.dataclasses import dataclass

@dataclass
class ServerConfig:
    host: str = "0.0.0.0"
    port: int = Field(8000, ge=1, le=65535)
    allowed_origins: list[str] = field(default_factory=lambda: ["*"])

@dataclass
class UIConfig:
    theme: str = "default"
    logo_url: str | None = None
    show_tool_output: bool = False

@dataclass
class LoggingConfig:
    level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    json_format: bool = True

@dataclass
class AppConfig:
    # `Any` here because each holds an instantiable DictConfig with _target_;
    # the concrete adapter validates its own kwargs in __post_init__/__init__.
    stt: Any
    llm: Any
    tts: Any
    input: Any
    audio: Any
    tools: Any
    server: ServerConfig = field(default_factory=ServerConfig)
    ui: UIConfig = field(default_factory=UIConfig)
    logging: LoggingConfig = field(default_factory=LoggingConfig)
    paths: dict[str, str] = field(default_factory=dict)
    scenarios_file: str = "config/system_prompts.yaml"
    predefined_answers_file: str = "config/predefined_answers.yaml"

# config/loader.py
from hydra import compose, initialize_config_dir
from hydra.core.config_store import ConfigStore
from omegaconf import OmegaConf

ConfigStore.instance().store(name="app_config_schema", node=AppConfig)

def load_config(overrides: list[str] | None = None) -> AppConfig:
    deployment = os.getenv("RVC_DEPLOYMENT")
    ov = list(overrides or [])
    if deployment:
        ov.append(f"+deployment={deployment}")
    with initialize_config_dir(config_dir=str(CONFIG_DIR), version_base="1.3"):
        cfg = compose(config_name="config", overrides=ov)
    OmegaConf.resolve(cfg)                       # resolve ${oc.env:...} now
    # Validate non-DI sections through Pydantic; DI sections validated on instantiate.
    return OmegaConf.to_object(cfg)              # → AppConfig (Pydantic-validated dataclass)

Each adapter validates its own construction args using a Pydantic model in its __init__, so instantiate(cfg.stt) fails immediately if, say, model is misspelled — the error names the exact field.

7.5 Secrets

# .env  (gitignored, never committed; loaded into env before Hydra resolves)
GROQ_API_KEY=gsk_...
OPENAI_API_KEY=sk-proj-...

• python-dotenv loads .env into os.environ at process start (before load_config).
• YAML references secrets only via ${oc.env:GROQ_API_KEY} — values never touch disk.
• Adapters store them as Pydantic SecretStr; structlog processors redact any SecretStr repr.
• Startup validation: if a selected adapter needs a key that resolves empty, construction raises a clear ConfigError and the service refuses to start.

7.6 Why this satisfies the abstraction goal

Requirement	Mechanism
Abstract interface per capability	abc.ABC ports in services/**/base.py (§5.2)
DI selects implementation from config	Hydra defaults list + instantiate(_target_=...) (§5.3)
Swap impl with zero code change	stt=groq → stt=openai in YAML or CLI
Per-site deployments	config/deployment/*.yaml presets
Type-safe, fail-fast	Pydantic structured configs + per-adapter arg validation
No secrets on disk	OmegaConf env resolvers + SecretStr

---

8. Service Layer Specifications

8.1 Audio Adapters — Recorder, AudioPlayer, DeviceProbe

The PoC's single AudioService is decomposed into three ports (§5.2). This makes "the start/stop-recording abstraction" first-class and independently swappable.

Recorder adapters — implement start_recording() / stop_recording() -> bytes:

class PyAudioRecorder(Recorder):
    def __init__(self, input_device: str | None, sample_rate: int, channels: int): ...
    def start_recording(self) -> None: ...      # opens callback stream, buffers frames
    def stop_recording(self) -> bytes: ...       # returns in-memory WAV

class ArecordRecorder(Recorder):
    """Subprocess fallback: /usr/bin/arecord -D plughw:{card},0 (full path for systemd)."""

class FallbackRecorder(Recorder):
    """Composite. Holds an ordered list of Recorder strategies; on failure of one,
    transparently advances to the next. Decided ONCE at start_recording, so a
    single utterance never mixes backends."""
    def __init__(self, strategies: list[Recorder], input_device: str | None, ...): ...

AudioPlayer adapter — the fallback player chain becomes an explicit ordered strategy list, not inline try/except:

class SubprocessPlayer(AudioPlayer):
    PLAYERS = [  # (binary, argv-builder); filtered by shutil.which() at init
        ("mpg123",  lambda f, card: ["mpg123", "-a", f"plughw:{card},0", str(f)]),
        ("ffplay",  lambda f, _:    ["ffplay", "-nodisp", "-autoexit", str(f)]),
        ("aplay",   lambda f, card: ["aplay", "-D", f"plughw:{card},0", str(f)]),
        ("paplay",  lambda f, _:    ["paplay", str(f)]),
        ("afplay",  lambda f, _:    ["afplay", str(f)]),
    ]
    async def play(self, audio: Path | bytes) -> None:
        # native async subprocess; raises AudioPlaybackError only if ALL fail
        for binary, build in self._available:
            proc = await asyncio.create_subprocess_exec(*build(path, self._card), ...)
            if await proc.wait() == 0: return
        raise AudioPlaybackError(...)

DeviceProbe adapter: LinuxDeviceProbe parses arecord -l / aplay -l / /proc/bus/input/devices. NullDeviceProbe (macOS/dev) reports all connected. Selected by config — no if platform == ... branches in business code.

8.2 UtteranceSource Adapters — VAD & Push-to-Talk

Both input modes implement the same UtteranceSource port (§5.2); the orchestrator is blind to which is active.

VadSource (continuous):

class VadSource(UtteranceSource):
    def __init__(self, energy_threshold: int, silence_duration: float,
                 pre_speech_buffer: float, min_speech_duration: float, sample_rate: int): ...
    async def start(self) -> None:
        self._task = asyncio.create_task(self._run())   # owns its loop
    async def _run(self) -> None:
        # native async pw-record subprocess; explicit Enum state machine:
        # LISTENING → RECORDING → COOLDOWN
        ...
    @property
    def idle_status(self) -> Literal["listening"]: return "listening"

- RMS via vectorized NumPy on int16 frames. - Pre-speech ring buffer: collections.deque(maxlen=N) (preserves leading syllables). - Cooldown gate: before invoking the utterance callback, check state.is_speaking or state.is_processing; if set, discard (robot must not hear itself).

PushToTalkSource (composes a KeyTrigger + a Recorder):

class PushToTalkSource(UtteranceSource):
    def __init__(self, trigger: KeyTrigger, recorder: Recorder): ...
    async def start(self) -> None:
        self._trigger.on_press(self._toggle)    # KEY_DOWN only
        self._trigger.start()
    @property
    def idle_status(self) -> Literal["idle"]: return "idle"

KeyTrigger adapters — EvdevTrigger (Linux headless) and PynputTrigger (display). Toggle-on-press-only: callback fires only on KEY_DOWN (event.value == 1), never on release — handles clickers that emit both on one physical press. evdev scans /dev/input/event* for EV_KEY devices; select() with 0.1 s timeout for clean shutdown.

8.3 LLM & STT Adapters (native async)

Performance correction vs. PoC: use the SDKs' native async clients (AsyncGroq, AsyncOpenAI) instead of wrapping sync calls in asyncio.to_thread. This removes thread-hop latency and gives true streaming on the event loop. All adapters share one httpx.AsyncClient (HTTP/2, keep-alive pool) so repeat API calls skip the TLS handshake.

GroqLLM:

class GroqLLM(LLMProvider):
    def __init__(self, api_key: SecretStr, model: str, max_tokens: int = 2048,
                 temperature: float = 0.7, http_client: httpx.AsyncClient | None = None):
        self._client = AsyncGroq(api_key=api_key.get_secret_value(), http_client=http_client)

    async def chat_stream(self, messages, tools) -> AsyncIterator[LLMChunk]:
        stream = await self._client.chat.completions.create(
            model=self._model, messages=[m.to_api() for m in messages],
            tools=tools or NOT_GIVEN, tool_choice="auto" if tools else NOT_GIVEN,
            max_tokens=self._max_tokens, temperature=self._temperature, stream=True,
        )
        async for chunk in stream:           # genuinely non-blocking
            yield self._parse_chunk(chunk)

GroqSTT (model whisper-large-v3-turbo) and OpenAISTT (whisper-1) both implement STTProvider.transcribe, awaiting client.audio.transcriptions.create(..., response_format="verbose_json") and returning TranscriptionResult(text, language).

8.4 TTS Adapters (native async)

OpenAITTS:

class OpenAITTS(TTSProvider):
    async def synthesize(self, text: str, language: str) -> Path:
        out = self._temp_dir / f"tts_{uuid4().hex}.mp3"   # UUID — safe under concurrency
        async with self._client.audio.speech.with_streaming_response.create(
            model=self._model, voice=self._voice, input=text,
        ) as resp:
            await resp.stream_to_file(out)                 # stream to disk, no thread hop
        return out

UUID filenames (never a fixed openai_tts.mp3) because chunks are synthesized concurrently (§Performance).

8.5 PiperTTS / EspeakTTS Adapters

VOICE_MAP = {"en": "en_US-lessac-medium", "de": "de_DE-thorsten-medium", "fr": "fr_FR-siwis-medium"}

async def synthesize(self, text: str, language: str) -> Path:
    voice = VOICE_MAP.get(language, VOICE_MAP["en"])
    output_path = self._temp_dir / f"tts_{uuid.uuid4().hex}.wav"
    process = await asyncio.create_subprocess_exec(
        "piper", "--model", voice, "--output_file", str(output_path),
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.DEVNULL,
        stderr=asyncio.subprocess.DEVNULL,
    )
    await process.communicate(input=text.encode())
    if process.returncode != 0:
        raise TTSSynthesisError(f"piper failed with code {process.returncode}")
    return output_path

8.6 McpToolBackend (implements ToolBackend)

Architecture: MCP's stdio client is async but the pipeline may call tools from contexts that need a stable loop, so the backend owns a dedicated event-loop thread and bridges via run_coroutine_threadsafe. It implements the ToolBackend port — the pipeline depends on the interface, and NullToolBackend is substituted when tools=none.

class McpToolBackend(ToolBackend):
    def __init__(self, servers: list[McpServerCfg], blocking_tools: list[str],
                 call_timeout_s: int = 30): ...

    @classmethod
    async def create(cls, cfg) -> "McpToolBackend":
        """Async factory: start loop thread, connect servers, discover tools."""

    def tool_definitions(self) -> list[ToolDefinition]:
        """OpenAI-format schemas; response_text auto-injected (optional param)."""

    async def call(self, tool_name: str, arguments: dict) -> str:
        fut = asyncio.run_coroutine_threadsafe(
            self._async_call(tool_name, arguments), self._loop)
        return await asyncio.wrap_future(fut)        # await without blocking the loop

    async def _async_call(self, tool_name: str, arguments: dict) -> str:
        session = self._resolve_session(tool_name)
        result = await asyncio.wait_for(
            session.call_tool(tool_name.split(".")[-1], arguments),
            timeout=self._timeout)
        return self._extract_content(result)

    def is_blocking(self, tool_name: str) -> bool:
        return tool_name in self._blocking_tools

    async def aclose(self) -> None: ...

Startup resilience: if a server fails to connect, log a warning and continue without it — the app starts with reduced tool capability rather than failing (Contract §18.10).

Tool injection: append optional response_text to every tool's parameters.properties ("The robot's spoken response to play while executing this action").

8.7 ToolDispatcher

Encapsulates all three tool execution modes. Depends only on the ToolBackend, TTSProvider, and AudioPlayer ports:

class ToolDispatcher:
    def __init__(
        self,
        tools: ToolBackend,
        tts: TTSProvider,
        player: AudioPlayer,
        playback: PlaybackQueue,
        emitter: EventEmitter,
        cfg: PipelineConfig,
    ): ...

    async def dispatch(
        self,
        tool_calls: list[ToolCall],
        language: str,
        state: AppState,
    ) -> DispatchResult:
        """
        Returns DispatchResult indicating whether predefined audio was played,
        or a blocking tool result was obtained.
        """

    async def _handle_predefined_answer(
        self, answer_id: str, language: str
    ) -> None:
        """Play pre-recorded audio, run wave gesture in parallel."""

    async def _handle_blocking_tool(
        self, tool_call: ToolCall, state: AppState
    ) -> str:
        """Play ack audio + run tool concurrently. Return tool result string."""

    async def _handle_nonblocking_tool(
        self, tool_call: ToolCall, tts_audio_path: Path | None
    ) -> None:
        """Fire-and-forget: gesture + audio in parallel."""

8.8 PlaybackQueue

Async queue consumed by a background coroutine. Depends on the AudioPlayer port:

class PlaybackQueue:
    def __init__(self, player: AudioPlayer, emitter: EventEmitter, state: AppState):
        self._queue: asyncio.Queue[Path | None] = asyncio.Queue()

    async def put(self, path: Path) -> None:
        await self._queue.put(path)

    async def clear(self) -> None:
        """Drain queue without playing."""
        while not self._queue.empty():
            try: self._queue.get_nowait()
            except asyncio.QueueEmpty: break

    async def run(self) -> None:
        """Background consumer. None sentinel = stop."""
        while True:
            item = await self._queue.get()
            if item is None:
                break
            state.is_speaking.set()
            await emitter.emit("status", StatusEvent(status="speaking"))
            try:
                await self._player.play(item)
            finally:
                item.unlink(missing_ok=True)  # cleanup temp file
        state.is_speaking.clear()
        await emitter.emit("status", StatusEvent(status="idle"))

    async def stop(self) -> None:
        await self._queue.put(None)

8.9 EventEmitter

Typed wrapper around Socket.IO emit. All WebSocket emissions go through this.

class EventEmitter:
    def __init__(self, sio: socketio.AsyncServer): ...

    async def emit_status(self, status: AppStatus) -> None:
        await self._sio.emit("status", {"status": status})

    async def emit_transcription(self, text: str, language: str) -> None:
        await self._sio.emit("transcription", {"text": text, "language": language})

    async def emit_response(self, text: str) -> None:
        await self._sio.emit("response", {"text": text})

    async def emit_response_chunk(self, text: str) -> None:
        await self._sio.emit("response_chunk", {"text": text})

    async def emit_error(self, message: str) -> None:
        await self._sio.emit("error", {"message": message})

    async def emit_tool_start(self, tool: str, args: dict) -> None:
        await self._sio.emit("tool_start", {"tool": tool, "args": args})

    async def emit_predefined_answer(self, answer_id: str) -> None:
        await self._sio.emit("predefined_answer", {"answer_id": answer_id})

8.10 Performance & Latency Architecture

The product is judged on time-to-first-audio (TTFA) — how long after the user stops speaking until the robot starts talking. The architecture is built to minimize it. The pipeline is a software pipeline in the literal sense: stages overlap rather than run sequentially.

1. Event loop & runtime - uvloop + httptools under Uvicorn — 2–4× faster loop/parsing than stock asyncio. - Native async SDK clients (AsyncGroq, AsyncOpenAI) — no to_thread hops on the hot path. - Shared httpx.AsyncClient with HTTP/2 and a keep-alive pool, injected into every API adapter. Connections to Groq/OpenAI stay warm, so each call skips DNS + TCP + TLS (~100–300 ms saved per call). - orjson for all (de)serialization (FastAPI ORJSONResponse, Socket.IO json module).

2. Streaming, stage-overlapped pipeline

LLM tokens ──► sentence splitter ──► TTS(sentence_1) ──► PlaybackQueue ──► speaker
                       │                    ▲                  │
                       └─ sentence_2 ──► TTS(sentence_2) ──────┘  (synthesized while #1 plays)

• TTS for sentence N+1 runs concurrently with playback of sentence N (asyncio.create_task per chunk, ordered by an asyncio.Queue). The user hears audio after the first sentence is synthesized, not the whole response.
• Sentence boundary detection (.!?\n) flushes a chunk to TTS the instant it completes.
• Robot gestures execute in parallel with speech (non-blocking tools) via concurrent tasks — never serialized behind audio.

3. TTFA budget (p95 target ≤ 2.5 s from speech-end to first audio)

Stage	Technique	Budget
Capture finalize	In-memory WAV, no temp file	~0 ms
STT	Groq turbo, warm connection, audio sent as bytes	≤ 800 ms
LLM first token	Streaming, warm connection, max_tokens capped	≤ 600 ms
First sentence assembled	Split on first boundary	+~200 ms
TTS first sentence	Short text, streaming-to-file, warm connection	≤ 700 ms
Playback start	Player process pre-resolved at startup	≤ 100 ms

4. Predefined-answer fast path - Predefined answers bypass TTS entirely — pre-rendered MP3 played directly. TTFA ≈ disk read + player spawn (≤ 150 ms). - All predefined MP3s pre-loaded into an in-memory LRU cache (bytes) at startup for the active scenario, so playback avoids a cold disk read.

5. Concurrency model & non-blocking discipline - No blocking I/O on the event loop. Subprocesses (pw-record, players, piper) use asyncio.create_subprocess_exec. - The only OS threads are the input-source listener (evdev/pw-record bridge) and the MCP event-loop thread; both hand work to the main loop via call_soon_threadsafe / wrap_future. - CPU-bound work (NumPy RMS in VAD) is trivially vectorized and runs in the VAD task without starving the loop (chunked, ~21 ms frames).

6. Frontend responsiveness - response_chunk events render tokens as they stream — perceived latency far below actual completion. - Vite production build, code-split, served as static assets (no SSR overhead).

7. ARM64 (Jetson) specifics - Prefer the apt python3-pyaudio binary over pip wheels. - uvloop and orjson ship aarch64 wheels — verify in uv.lock. - Pin a single Uvicorn worker (the app is a single stateful process; see §15.1).

8. Measured continuously - Every stage emits a duration_ms structured log (§13.3); a pipeline.ttfa_ms metric is logged per interaction so regressions are caught from logs alone.

---

9. API Specification

All responses use consistent envelope:

{ "data": {...}, "error": null }
{ "data": null, "error": { "code": "INVALID_SCENARIO", "message": "..." } }

GET /api/scenarios

Returns available scenarios and current selection.

{
  "data": {
    "scenarios": ["Belimed", "ZHAW", "Funny Chat"],
    "current": "Belimed"
  }
}

POST /api/scenario

Switch active scenario.

Request:  { "scenario": "ZHAW" }
Response: { "data": { "scenario": "ZHAW" } }
Error 400: { "data": null, "error": { "code": "INVALID_SCENARIO", "message": "..." } }
Error 409: { "data": null, "error": { "code": "PIPELINE_BUSY", "message": "Cannot switch while processing" } }

GET /api/config

Returns runtime UI configuration (no secrets).

{
  "data": {
    "stt_provider": "groq",
    "tts_provider": "openai",
    "llm_model": "llama-3.3-70b-versatile",
    "debug_enabled": false,
    "show_tool_output": true,
    "theme": "belimed",
    "logo_url": "https://..."
  }
}

GET /api/device_status

Check peripheral connectivity.

{
  "data": {
    "microphone": { "connected": true, "name": "USB Composite Device" },
    "speaker":    { "connected": true, "name": "UACDemoV1.0" },
    "keyboard":   { "connected": true, "name": "clicker detected" }
  }
}

On non-Linux systems, return all as connected: true (no arecord/aplay available).

GET /api/health

Liveness and readiness probe (for monitoring/systemd).

{
  "data": {
    "status": "ok",
    "mcp_connected": true,
    "stt_provider": "groq",
    "uptime_seconds": 3600
  }
}

POST /api/admin/restart

Restart the systemd service. Requires systemctl passwordless sudo configured.

Response: { "data": { "message": "Restart initiated" } }

Implementation: Run subprocess.Popen(["sudo", "systemctl", "restart", "robot-voice-chat"]). Always return 200 (service kills itself). Do not wait.

Rate limit: 1 request per 30 seconds.

---

10. WebSocket Protocol

Server: python-socketio with AsyncServer. Transport: WebSocket + polling fallback. Namespace: default (/).

Client → Server Events

Event	Payload	Description
start_recording	{}	Begin audio capture. Ignored if already recording/processing/speaking.
stop_recording	{}	End audio capture. Triggers pipeline. Ignored if not recording.
send_message	{ text: string, language?: string }	(Chat modality, §5.8) Submit a typed user turn. Routed to ChatInputChannel.submit(). Emits an error with code CHAT_DISABLED if the chat modality is off. Ignored if text is blank. language defaults to "en".
change_scenario	{ name: string }	Switch the active scenario. Emits scenario_changed, or an error with code UNKNOWN_SCENARIO.
stop_pipeline	{}	Barge-in: interrupt current speech without aborting the in-flight LLM call.

Server → Client Events

Event	Payload	Description
status	{ status: AppStatus }	State machine update
transcription	{ text: string, language: string }	STT result
response	{ text: string }	Complete LLM response
response_chunk	{ text: string }	Streaming token (incremental)
predefined_answer	{ answer_id: string }	Predefined audio selected
tool_start	{ tool: string, args: object }	Tool execution beginning
tool_result	{ tool: string, result: string }	Tool execution result
error	{ message: string, code?: string }	Error occurred

AppStatus values: "idle" | "recording" | "processing" | "speaking" | "listening"

• "listening" = VAD mode, actively monitoring for speech
• "idle" = PTT mode, waiting for key press
• "recording" = audio capture in progress
• "processing" = STT + LLM in progress
• "speaking" = TTS audio playback in progress

---

11. Data Models

All backend models in Pydantic v2.

# models/chat.py

class Message(BaseModel):
    role: Literal["system", "user", "assistant", "tool"]
    content: str

class ToolCallFunction(BaseModel):
    name: str
    arguments: str  # JSON string

class ToolCall(BaseModel):
    id: str
    type: Literal["function"] = "function"
    function: ToolCallFunction

    def parsed_arguments(self) -> dict:
        try:
            return json.loads(self.function.arguments)
        except json.JSONDecodeError:
            return {}

class ToolDefinition(BaseModel):
    type: Literal["function"] = "function"
    function: dict  # OpenAI function schema format

class LLMChunk(BaseModel):
    content: str | None = None
    tool_call_delta: ToolCallDelta | None = None
    finish_reason: str | None = None

class LLMResponse(BaseModel):
    content: str | None
    tool_calls: list[ToolCall] = []

# models/scenario.py

class PredefinedAnswer(BaseModel):
    id: str
    description: str
    text_en: str
    text_de: str | None = None
    text_fr: str | None = None

    def text_for_language(self, lang: str) -> str:
        return getattr(self, f"text_{lang}", None) or self.text_en

class Scenario(BaseModel):
    name: str
    system_prompt: str

# models/audio.py

class AudioDevice(BaseModel):
    index: int
    name: str
    max_input_channels: int
    max_output_channels: int
    default_sample_rate: float

class TranscriptionResult(BaseModel):
    text: str
    language: str  # full name: "english", "german", etc.

    @property
    def language_code(self) -> str:
        mapping = {"english": "en", "german": "de", "french": "fr", "italian": "it"}
        return mapping.get(self.language.lower(), "en")

---

12. Security Architecture

12.1 Secrets Management

• API keys only in environment variables or .env file (which is gitignored)
• .env.example committed with placeholder values
• Settings startup validation: if required key missing, raise ValueError with clear message before service starts
• SecretStr type in Pydantic prevents accidental logging of key values

12.2 Input Validation

All HTTP request bodies validated by Pydantic. Scenario names validated against known list:

@router.post("/api/scenario")
async def set_scenario(
    body: SetScenarioRequest,
    loader: ScenarioLoader = Depends(get_scenario_loader),
) -> ScenarioResponse:
    valid = [s.name for s in loader.scenarios]
    if body.scenario not in valid:
        raise HTTPException(status_code=400, detail={"code": "INVALID_SCENARIO"})

12.3 CORS Configuration

Configure explicitly for production. Do not use * in production:

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.server.allowed_origins,  # e.g., ["http://192.168.1.100:8000"]
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type"],
)

For Socket.IO: cors_allowed_origins=settings.server.allowed_origins

Default for development: ["*"]. Production: restrict to robot's IP.

12.4 Rate Limiting

Use slowapi:

from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)

@router.post("/api/admin/restart")
@limiter.limit("1/30seconds")
async def restart(...): ...

@router.get("/api/device_status")
@limiter.limit("12/minute")  # client polls every 10s
async def device_status(...): ...

12.5 Subprocess Security

All subprocess calls use lists (not shell strings) to prevent injection:

# CORRECT
subprocess.run(["/usr/bin/arecord", "-D", f"plughw:{card},0", "-f", "cd"], ...)

# NEVER do this
subprocess.run(f"arecord -D plughw:{card},0 -f cd", shell=True)  # vulnerable

Validate ALSA card numbers are integers before use.

12.6 Temporary File Security

TTS temporary files: - Written to audio/temp/ directory - UUID-based names (no predictable paths) - Deleted immediately after playback completes - Directory created at startup with appropriate permissions (700)

12.7 Flask SECRET_KEY Equivalent

FastAPI does not use a session secret by default. If sessions are added, use a randomly-generated secret from environment:

APP_SECRET_KEY=<random 64-char hex>  # in .env

12.8 Admin Panel Authentication

The /admin page and every /api/admin/* endpoint (except /login) are gated behind a password set in the environment. Admin is disabled by default — when the password is unset, every admin endpoint returns HTTP 503 and the UI shows that admin is not configured. There are no default credentials.

Configuration (backend/.env):

# Argon2 hash. Generate with:
#   uv run python -m robot_voice_chat.scripts.hash_password
# Plaintext ADMIN_PASSWORD is NOT supported.
ADMIN_PASSWORD_HASH=$argon2id$v=19$m=65536,t=3,p=4$...

# HMAC key for signing session tokens. Optional; if absent, a fresh random key
# is generated at startup (tokens then invalidate on every restart). Set
# persistently in production:
#   python -c "import secrets; print(secrets.token_urlsafe(32))"
ADMIN_SECRET_KEY=<random url-safe string>

Token model — stateless, stdlib-only:

• Login (POST /api/admin/login) verifies the password with secrets.compare_digest (constant-time) and returns a bearer token. Rate-limited to 5 req/min/IP via slowapi.
• Token format: base64url(JSON{v, exp}).base64url(HMAC-SHA256(secret, payload)). 8-hour expiry by default.
• Verification uses hmac.compare_digest and checks payload version + expiry.
• Frontend stores the token in sessionStorage (clears on tab close) and sends it as Authorization: Bearer <token> on every admin request. A 401 response clears the cached token automatically.
• The reusable require_admin FastAPI dependency gates any new admin endpoint with a single dependencies=[Depends(require_admin)] declaration.

Transport — bearer tokens travel in headers, so the admin panel must be served over TLS in production (reverse-proxy with Caddy / nginx / Traefik). The systemd unit binds to localhost-only by default; expose the proxy.

---

13. Observability & Logging

13.1 Structured Logging with structlog

import structlog

logger = structlog.get_logger()

# Configure at startup
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer() if settings.logging.json_format
        else structlog.dev.ConsoleRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(
        getattr(logging, settings.logging.level)
    ),
)

Structured log fields (all logs should include relevant subset):

log = logger.bind(request_id=request_id, scenario=scenario_name)
log.info("transcription_complete", text=result.text, language=result.language_code, duration_ms=elapsed)
log.info("llm_response_complete", chars=len(response), tool_calls=len(tool_calls), duration_ms=elapsed)
log.info("tts_synthesized", provider=provider, language=lang, chars=len(text), duration_ms=elapsed)
log.error("mcp_tool_failed", tool=tool_name, error=str(e), duration_ms=elapsed)

13.2 Request ID Tracing

Inject a request/session ID into the structlog context at the start of each pipeline run:

import uuid
structlog.contextvars.bind_contextvars(pipeline_run_id=str(uuid.uuid4()))
# ... run pipeline ...
structlog.contextvars.clear_contextvars()

This allows correlating all log lines for a single voice interaction.

13.3 Timing

Log duration for each major stage: - Audio recording duration - STT API latency - LLM first-token latency (time to first chunk) - LLM total latency - TTS latency per chunk - Playback duration - MCP tool call latency

Use a context manager:

@contextmanager
def timed(log, event: str, **ctx):
    start = time.perf_counter()
    yield
    log.info(event, duration_ms=int((time.perf_counter() - start) * 1000), **ctx)

13.4 Health Endpoint

GET /api/health returns structured status for uptime monitoring and systemd ExecStartPost checks.

13.5 Log Rotation

Configure in deploy/robot-voice-chat.service:

StandardOutput=journal
StandardError=journal
SyslogIdentifier=robot-voice-chat

Use journald for log management (rotation built-in). For file-based logs, use logrotate.

---

14. Testing Strategy

14.1 Unit Tests

tests/unit/test_config.py: - load_config() composes defaults without error - Missing required secret env var → adapter construction raises ConfigError at startup - stt=openai override selects the OpenAISTT _target_ (assert instantiate builds the right class) - Deployment preset (+deployment=belimed) applies all overrides - Invalid value (e.g. server.port=99999) fails Pydantic validation with a precise field error

tests/unit/test_audio_adapters.py: - FallbackRecorder advances to next strategy when first raises; never mixes within one utterance - SubprocessPlayer filters by shutil.which, tries chain in order, raises only if all fail - Card number parsing from arecord -l output

tests/unit/test_vad_source.py: - RMS computation correct for known signals - State machine transitions: LISTENING → RECORDING on high RMS - RECORDING → LISTENING on silence timeout - Callback not invoked when is_speaking set - Pre-speech buffer included in output WAV

tests/unit/test_tool_dispatcher.py: - Predefined answer: correct file path constructed per language - Fallback to English when translation missing - Blocking tool: clears audio queue before executing - Non-blocking tool: does not block pipeline

tests/unit/test_orchestrator.py: - Empty transcription raises error, emits error event - LLM no-content fallback triggers non-streaming call - Streaming sentence splits at correct boundaries - Language mapping: "german" → "de"

14.2 Integration Tests

tests/integration/test_api_routes.py: Use httpx.AsyncClient with ASGITransport:

@pytest.fixture
async def client(app):
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

async def test_get_scenarios(client):
    response = await client.get("/api/scenarios")
    assert response.status_code == 200
    assert "scenarios" in response.json()["data"]

async def test_set_invalid_scenario(client):
    response = await client.post("/api/scenario", json={"scenario": "DoesNotExist"})
    assert response.status_code == 400

tests/integration/test_websocket.py: Use python-socketio test client:

async def test_status_on_connect(sio_client):
    events = []
    sio_client.on("status", lambda data: events.append(data))
    await sio_client.connect("http://localhost:8000")
    assert events[0]["status"] in ("idle", "listening")

14.3 Test Configuration

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
filterwarnings = ["error"]

[tool.coverage.run]
source = ["src/robot_voice_chat"]
omit = ["*/tests/*"]

[tool.coverage.report]
fail_under = 80

14.4 Mocking Strategy

Because every capability is an abc.ABC port (§5.2), tests substitute fakes at the interface boundary — no patching of internals. - STT/LLM/TTS/ToolBackend: hand the orchestrator in-memory fakes implementing the port (FakeSTT, FakeLLM yielding scripted chunks, FakeTTS writing a silent WAV, FakeToolBackend). - Recorder/AudioPlayer/UtteranceSource: FakeRecorder returns canned WAV bytes; RecordingPlayer records calls instead of playing. - API routes: app.dependency_overrides[get_device_probe] = lambda: FakeProbe(). - HTTP-level (when testing real adapters): respx mocks the httpx transport so SDK code runs unmodified. - Subprocess adapters: monkeypatch asyncio.create_subprocess_exec. - Hydra: build the graph from a test config dir with initialize_config_dir, overriding groups to fake _target_s where useful.

---

15. Deployment Architecture

15.1 systemd Service

# deploy/robot-voice-chat.service
[Unit]
Description=Robot Voice Chat
After=network.target sound.target

[Service]
Type=exec
User=robot
Group=robot
WorkingDirectory=/opt/robot-voice-chat
EnvironmentFile=/opt/robot-voice-chat/.env
ExecStart=/opt/robot-voice-chat/.venv/bin/uvicorn \
    robot_voice_chat.main:socket_app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 1 \
    --no-access-log
Restart=on-failure
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=robot-voice-chat

# Security hardening
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/robot-voice-chat/audio

[Install]
WantedBy=multi-user.target

Critical: --workers 1. Multiple workers would create multiple instances of all services (audio, keyboard, MCP) which is incorrect. The application is single-process by design.

15.2 Required System Permissions

User robot must be in groups: - audio: ALSA audio devices - input: /dev/input/event* keyboard devices

sudo usermod -a -G audio,input robot

Passwordless sudo for service restart:

# /etc/sudoers.d/robot-voice-chat
robot ALL=(ALL) NOPASSWD: /bin/systemctl restart robot-voice-chat

15.3 Dependency Installation

# System deps (Ubuntu/Jetson)
sudo apt-get install -y \
    portaudio19-dev python3-pyaudio \
    libasound2-dev ffmpeg mpg123 espeak \
    pipewire libgirepository1.0-dev

# Python deps via uv (resolves fastapi, uvicorn[standard], hydra-core, omegaconf,
# pydantic, structlog, groq, openai, httpx, orjson, sounddevice, mcp, slowapi, numpy...)
cd backend
uv sync --frozen          # installs from uv.lock
# Piper is optional (no ARM64 wheel): uv sync --extra piper  (x86 dev only)

# Frontend build
cd frontend
pnpm install
pnpm build  # outputs to backend/src/robot_voice_chat/static

15.4 File Paths (Production)

/opt/robot-voice-chat/          # Application root
├── .env                        # Secrets (owned by robot:robot, chmod 600)
├── .venv/                      # Python virtual environment (uv managed)
├── config/                     # YAML configs (version controlled or deployed)
├── audio/
│   ├── predefined/             # Pre-recorded MP3s
│   └── temp/                   # Ephemeral TTS files (chmod 700)
└── src/                        # Application source
    └── robot_voice_chat/
        └── static/             # Built frontend assets

15.5 Startup Verification Script

#!/bin/bash
# scripts/verify_setup.py (run before first start)
# Checks: .env exists, API keys valid format, audio devices present, Python version

---

16. Non-Functional Requirements

16.1 Performance

See §8.10 for the latency architecture. Targets (the how is the streaming, stage-overlapped pipeline + warm async clients):

Metric	Target	Notes
STT latency	< 800 ms	Groq turbo, warm HTTP/2 connection
LLM first token	< 600 ms	streaming, warm connection
TTS first sentence	< 700 ms	streaming-to-file, warm connection
Time-to-first-audio (speech-end → first audio)	< 2.5 s p95	overlapped pipeline; predefined path < 150 ms
Memory (idle)	< 200 MB	Jetson constraint
CPU (idle)	< 5%	Jetson constraint

16.2 Reliability

• Application must restart automatically on crash (systemd Restart=on-failure)
• MCP server disconnection must not crash the application (log warning, continue without tools)
• STT/LLM/TTS API errors must emit error WebSocket event, return to idle state
• No global state that could deadlock: use timeouts on all blocking operations

16.3 Maintainability

• All service boundaries defined by abstract ports; the application/pipeline layer never imports a concrete adapter (enforce via an import-linter contract in CI)
• Implementations are selected exclusively through Hydra config — no if provider == ... ladders anywhere
• No hardcoded values outside the config tree (documented business rules like the 30 s MCP timeout live as a named config field with a comment)
• ruff check src/ zero warnings; mypy --strict clean on all source
• Code coverage ≥ 80% on all service/pipeline modules

16.4 Compatibility

Platform	Status
NVIDIA Jetson Orin NX (ARM64, Ubuntu 22.04)	Primary target
macOS 13+ (Apple Silicon or Intel)	Development
Ubuntu 22.04 x86_64	Secondary target

ARM64-specific: - Do NOT include piper-tts in default requirements (no ARM64 wheel). Include in optional extras: [piper] - Test with PyAudio ARM64 binary from apt (python3-pyaudio) rather than pip wheel

---

18. Key Behavioral Contracts

Critical behaviors the brain preserves. Keep them intact when changing pipeline or tool-dispatch code.

18.1 Toggle-on-Press-Only (Keyboard Input)

Clickers that physically send KEY_DOWN + KEY_UP on a single button press must trigger the pipeline exactly once (on KEY_DOWN). Never act on KEY_UP events from the hardware keyboard service. The frontend browser PTT (space bar) is separate — it uses keydown / keyup to implement hold-to-talk.

18.2 Pre-Speech Audio Buffer (VAD)

VAD must retain the last pre_speech_buffer seconds of audio before the energy threshold fires. Without this, the first syllable(s) of every utterance are clipped. Use a collections.deque(maxlen=N) ring buffer.

18.3 Streaming Sentence Splitting

TTS synthesis must begin before the LLM finishes. Split on ., !, ?, \n. Accumulate a sentence buffer. When a boundary character is seen AND the buffer has at least one non-whitespace character, synthesize immediately and push to playback queue. Do not wait for the full LLM response.

18.4 Predefined Answer Audio Fallback

If audio/predefined/{answer_id}_{lang}.mp3 does not exist, try audio/predefined/{answer_id}_en.mp3 before raising an error. Log a warning on fallback.

18.5 VAD Cooldown During Speaking

While is_speaking or is_processing is set, VAD callbacks must be silently discarded. The robot must not hear its own TTS output and start a new pipeline run.

18.6 MCP Tool Argument Cleaning

Before calling an MCP server, remove the response_text key from the arguments dict (it was injected for internal parallel-execution signaling and is not a real tool parameter).

18.7 Non-Blocking MCP Tools Are Fire-and-Forget

MCP tool results for non-blocking tools are not fed back to the LLM. The pipeline does not wait for them to complete. Log errors but do not raise.

18.8 Blocking Tools Interrupt TTS Queue

When a blocking tool is triggered, immediately call audio_queue.clear() to discard any already-queued TTS files. Then synthesize a short acknowledgment, play it, execute the tool, make a follow-up LLM call, and synthesize the new response. This ensures the robot doesn't finish playing a stale response while the camera is scanning.

18.9 LLM Streaming Fallback

If LLM streaming returns neither content nor tool calls, make one additional non-streaming call. This covers edge cases where the streaming endpoint returns an empty response.

18.10 MCP Server Startup Resilience

If any MCP server fails to connect at startup, log a warning but continue. The application should start successfully with reduced tool capability rather than failing entirely.

---

19. Production Hardening

The brain ships with the production-hardening measures below, on top of the clean ports/adapters core, typed Hydra+Pydantic config, structured logging, admin panel, and metrics store. Each item is independent.

Most §19.1–§19.8 items are now landed; the few deferred sub-items are called out inline. See the code repo's CHANGELOG.md for a summary of what shipped.

19.1 Security Hardening

• TLS + reverse-proxy guidance. systemd unit binds uvicorn to 127.0.0.1. Sample Caddy and nginx configs ship as deploy/Caddyfile.example and deploy/nginx.conf.example; README has a "TLS via a reverse proxy" section.
• Narrow CORS allowlist. server.allowed_origins is enforced via FastAPI CORSMiddleware. New server.environment: dev | production field; in production, startup refuses * or an empty allowlist. The belimed preset shows the production setup.
• Hashed admin password. ADMIN_PASSWORD_HASH (Argon2 via argon2-cffi) is the only supported form. Plaintext ADMIN_PASSWORD is rejected at startup with a guidance log line. Operator helper: uv run python -m robot_voice_chat.scripts.hash_password.
• Token revocation watermark. Issued tokens carry an iat; a persistent SessionStore keeps a min_iat watermark. POST /api/admin/logout-all bumps the watermark, invalidating every outstanding token at once.
• Security headers middleware. SecurityHeadersMiddleware emits HSTS, X-Content-Type-Options: nosniff, X-Frame-Options: DENY, Referrer-Policy: strict-origin-when-cross-origin, and a default CSP covering the SPA, Google Fonts, the Swagger CDN, and Socket.IO. Override per deployment via server.csp_policy.
• HTTPS-only cookies (if cookies ever return). Documented in the README "TLS via a reverse proxy" section — bearer-in-sessionStorage accepts XSS exposure to avoid CSRF; any future cookie must be Secure, HttpOnly, SameSite=Strict.

19.3 Observability Completeness

• Request/turn IDs via structlog.contextvars. RequestIdMiddleware binds request_id (UUID4 or inbound X-Request-ID) for the request's life and echoes it on the response. process_turn binds a fresh turn_id, so STT → LLM → TTS log lines for one user message correlate trivially.
• OpenTelemetry traces. Flag-gated by observability.otel_enabled; SDK lazy-imported (install with uv sync --extra observability). FastAPIInstrumentor wraps HTTP; the orchestrator wraps each turn in a pipeline.turn span. OTLP gRPC exporter; endpoint configurable.
• Prometheus /api/metrics endpoint. Flag-gated by observability.prometheus_enabled. Exposes the default prometheus_client registry (process + GC counters); domain metrics still land in the SQLite sink for the admin Stats page.
• Metrics retention job. Startup prune in lifespan; observability.retention_days (default 30) controls the cut-off. Runs VACUUM after a non-zero delete. Config-snapshot rows are exempt — the audit trail must survive retention.
• External error tracking hook (Sentry). Strictly opt-in via observability.sentry_dsn. sentry-sdk lazy-imported with FastAPI + Starlette integrations; environment + traces sample rate configurable.
• Liveness vs readiness probes. /api/livez returns 200 whenever the process is up; /api/readyz returns 503 when any required dependency (llm, stt/tts when their modality is on, MCP tools when it exposes is_connected()) is unavailable. /api/health preserved for back-compat.

19.4 Reliability & Resilience

• Retries with backoff on STT/LLM/TTS. services/resilience.py provides call_with_retries() wrapping calls in tenacity with 3 attempts and 0.25 s → 1 s → 4 s exponential backoff. Retries on connection errors and 5xx; never on 4xx. Each retry emits provider_retry at warning level so the request_id binding correlates.
• Explicit per-provider timeouts. STT/LLM/TTS adapters accept request_timeout_s (default 30 s) and thread it into the underlying SDK client (Groq / OpenAI). YAML defaults updated in config/{stt,llm,tts}/*.yaml.
• Circuit breaker per external provider. (Deferred — needs additional architecture; the retry + timeout combo already prevents the simplest cascading-slowdown failure mode.)
• Model fallback chain. (Deferred — needs a FallbackProvider adapter that takes providers: list[Provider] and races / falls back. Tracking separately.)
• Graceful shutdown. Lifespan now polls state.is_processing for up to observability.shutdown_drain_timeout_s (default 10 s) before tearing down providers. Logs shutdown_drain_started / shutdown_drain_complete / shutdown_drain_timeout.
• Document single-instance design. Documented under §16 + the new "Operations / Scaling" subsection of the README: AppState and PlaybackQueue are in-process; horizontal scaling requires a shared message bus.

19.5 Release & Deployment Pipeline

• CI on every PR. .github/workflows/ci.yml runs backend (pytest, ruff check, ruff format --check, mypy, pip-audit) and frontend (npm ci, typecheck, build, npm audit) jobs. Concurrency group cancels superseded runs.
• Dockerfile + docker-compose. Multi-stage Dockerfile (Node builder → python:3.13-slim runtime via uv, non-root user). docker-compose.yml brings up the backend plus a placeholder MCP container.
• Automated dependency hygiene. .github/dependabot.yml schedules weekly grouped PRs for the uv ecosystem (prod + dev), npm, and monthly for GitHub Actions.
• Semantic versioning + changelog. CHANGELOG.md adopts the Keep-a-Changelog format; SemVer applies to future releases.
• SBOM + SAST/SCA in CI. pip-audit (backend) + npm audit (frontend) run in CI; non-blocking until findings are addressed. SBOM generation + Trivy can be added once the CI is green by default.
• dev / staging / prod progression. (Partial — documented in README; full ladder doc deferred.)
• Pre-commit hooks. .pre-commit-config.yaml runs trailing-whitespace, ruff (autofix), ruff-format, and a local mypy hook on staged files.

19.6 AI / LLM Governance

• Per-turn token tracking. LLMUsage model added to models/chat.py; GroqLLM extracts usage from both streaming-final and non-streaming responses. The orchestrator emits an llm_usage log event with prompt_tokens / completion_tokens / total_tokens, which lands in the SQLite events table via the structlog -> metrics sink.
• Budget alerts. (Deferred — token tracking is in place but the budget-threshold + webhook layer is a follow-up. Track separately.)
• Prompt-injection defenses. The orchestrator wraps user input in <user_input>…</user_input> delimiters and appends a hardening sentence to the system prompt instructing the model to ignore any role-override attempts inside that block. Cheap, well-known mitigation against the obvious "ignore previous instructions" attack.
• Output content safety filter port. services/safety/ introduces an abstract ContentSafetyFilter with a NullContentSafetyFilter default. The orchestrator runs the assistant response through filter.check() before emit/speak; on a block, replaces with a safe fallback and logs safety_filter_blocked. Drop-in for an Azure / OpenAI / custom adapter without pipeline edits.
• Eval framework for prompt regressions. (Deferred — expensive to maintain; suggest gating behind a manual workflow trigger once datasets exist.)
• Conversation memory. (Deferred — optional per the spec, requires session identity story.)

19.7 Data & Privacy

• GDPR / CCPA affordances. (Deferred — the deployment has no identity model today; export/delete endpoints become meaningful only once sessions are tied to identities. The "Data retention by class" table below documents what is retained in the meantime.)
• Document retention per data class. Codified as a table in the README under "Data retention by class" + in §16:

Class	Where	Retention
Logs	In-memory ring buffer	Last log_buffer_capacity records (default 500). Lost on restart.
Metrics events	SQLite events table	observability.retention_days (default 30 days). Startup prune + VACUUM.
Audio temp files	audio/temp/	Deleted after playback by the TTS adapter.
Admin sessions	data/admin_session.json	min_iat watermark only; bumped by /admin/logout-all.

19.8 Frontend Production Readiness

• ErrorBoundary components. components/ErrorBoundary.tsx (class component) catches render-time errors and shows a "Something went wrong" card with "Try again" + "Reload" actions. Wraps the app at the top level and again per-admin-route (so an admin-page bug never takes down the chat UI).
• Code splitting. Every admin sub-page is React.lazy()-loaded behind Suspense. The chat-only initial bundle drops from ~817 KB to ~428 KB (recharts ships in its own AdminStats chunk that only loads when the user opens the stats page).
• Accessibility audit (partial). ChatComposer's icon-only send button already carries aria-label="Send message". (Deferred: a full axe-core sweep + screen-reader announcements via aria-live. Tracking separately.)
• i18n with locale switcher. (Deferred — out of scope for this batch.)

19.10 Documentation & Developer Experience

• Published docs site. (Deferred — mkdocs-material / Sphinx is straightforward to add later; the spec + README + CHANGELOG cover the immediate need.)