Skip to content
AuthorPascal DateJuly 14, 2026 Rev1.0

Coding guidelines

The conventions every binabik codebase follows. They're distilled from how the brain (robot-voice-chat) is built; the same patterns apply to the MCP servers and other services. For the full detail behind any rule here, see the Brain architecture reference.

The one rule that drives all the others

Program to ports, not implementations. Every capability that touches hardware, the network, or has more than one variant is an abstract port; the concrete adapter is chosen at composition time from config. Everything below follows from that.

Language, tooling & stack

  • Python 3.13, managed with uv (lockfile committed). Node with pnpm.
  • ruff for lint + format; mypy --strict for types. Both must be clean — zero warnings — and are enforced in CI. Pre-commit hooks run them on staged files.
  • Backend framework: FastAPI (async-native) + Uvicorn/uvloop, Socket.IO (python-socketio, ASGI) for realtime, orjson responses.
  • Config: Hydra (composition + DI) + Pydantic v2 (validation). See Configuration.
  • External calls use native async SDK clients (AsyncGroq, AsyncOpenAI) over a shared httpx.AsyncClient (HTTP/2, keep-alive) — never thread-pool-wrapped sync clients.
  • Logging: structlog (JSON in production).
  • Frontend: Vite + React + TypeScript, Tailwind + shadcn/ui, Zustand for client state, TanStack Query for server state, socket.io-client.

Don't add

No ORM/database (the app is stateless), no message broker, and no third-party DI container — Hydra instantiate is the DI container and FastAPI Depends() covers request scope. Adding another is redundant.

Architecture: ports & adapters

The backend is strictly layered (hexagonal); dependencies point inward only:

Transport   api/ (REST)      websocket/ (Socket.IO)        ← adapters in
Application pipeline/ (orchestrator, dispatcher, queue)    ← use cases
Domain      services/**/base.py  (abstract ports)          ← interfaces
Adapters    services/**/*.py     (concrete implementations)← adapters out
  • The application/pipeline layer imports only abstract ports (base.py), never a concrete adapter. This is enforced in CI with an import-linter contract.
  • Define a port as abc.ABC with @abstractmethods when it is instantiated via Hydra (_target_ needs a concrete class). Use Protocol only for purely structural, never-instantiated contracts.
  • Ports today: Recorder, AudioPlayer, DeviceProbe, UtteranceSource, STTProvider, LLMProvider, TTSProvider, ToolBackend, plus the modality ports InputChannel / SpeechOutput. Provide a Null* adapter (e.g. NullToolBackend) for the "disabled" case rather than sprinkling if enabled.
  • Fallback chains (recorder, audio player) are explicit composite adapters (Chain of Responsibility), not inline try/except.
  • One brain build serves every robot. Anything robot-, task-, or customer-specific is a registered service or a config value — never a fork.

Dependency injection & composition

  • A single CompositionRoot.build() assembles the object graph once at startup; adapters declare their dependencies in their constructors — no globals, no service locator.
  • instantiate(cfg.stt) reads the _target_ from the composed config and builds the right class. Switching an implementation is a one-line config change — there are no if provider == ... factory ladders anywhere.
  • All initialization happens in FastAPI's lifespan, never at import time. Load Hydra config with the Compose API (not @hydra.main) so the process stays a clean long-running service.
  • Routes receive ports, not concretes, via thin Depends() providers that pull singletons from app.state. Tests override them with app.dependency_overrides[...].

Configuration

  • Two jobs, split deliberately: Hydra config groups + instantiate select and construct implementations; Pydantic v2 validates and types the composed result so a bad value fails at startup, never at request time.
  • One config group (a directory under config/) per port; one YAML per selectable implementation, each carrying its _target_.
  • No secrets in YAML or version control. API keys come from the environment (via .env, gitignored; .env.example committed), are injected through OmegaConf env resolvers, and are typed SecretStr.
  • No hardcoded values outside the config tree. Business rules (e.g. the 30 s MCP timeout) live as named config fields with a comment.
  • Whole-deployment presets live under config/deployment/ and compose everything for one site.

Async & concurrency

  • The hot path is fully async on one event loop. Anything running in a background thread (audio capture, VAD) bridges onto the loop with loop.call_soon_threadsafe / asyncio.run_coroutine_threadsafe.
  • Shared mutable state (AppState) protects every field with a lock or threading.Event; expose it through typed accessors, not raw attributes.
  • Put a timeout on every blocking/external operation. No unbounded waits.

Security

  • Secrets only in env / .env; validate required keys at startup with a clear error; SecretStr prevents accidental logging.
  • Validate all HTTP bodies with Pydantic; validate enums/names against the known set.
  • CORS: explicit allowlist in production — the app refuses * when server.environment == production.
  • Subprocess calls use argument lists, never shell=True; validate any numeric inputs (e.g. ALSA card numbers) before use.
  • TTS temp files: UUID names under a 700 dir, deleted right after playback.
  • Admin endpoints are disabled unless an Argon2 ADMIN_PASSWORD_HASH is set (no plaintext, no defaults); bearer tokens are HMAC-signed and constant-time compared. Serve admin over TLS (reverse proxy).
  • A SecurityHeadersMiddleware sets HSTS, X-Content-Type-Options, X-Frame-Options, Referrer-Policy, and a CSP.

Logging & observability

  • Log through structlog with event names + structured fields (log.info("tts_synthesized", provider=..., duration_ms=...)), not f-string prose.
  • Bind a request_id per request and a turn_id per pipeline run via structlog.contextvars, so all lines for one interaction correlate.
  • Time every major stage (STT, LLM first-token, LLM total, TTS/chunk, tool call) and log duration_ms.
  • Health/monitoring: /api/livez (process up), /api/readyz (503 if a required dependency is down). OpenTelemetry, Prometheus, and Sentry are flag-gated and lazy-imported — off by default.

Testing

  • pytest + pytest-asyncio (asyncio_mode = "auto"), filterwarnings = ["error"].
  • Because every capability is a port, substitute fakes at the interface boundary — no patching of internals. FakeSTT, FakeLLM (scripted chunks), FakeTTS, FakeToolBackend, etc.
  • Mock HTTP at the transport with respx so SDK code runs unmodified; monkeypatch asyncio.create_subprocess_exec for subprocess adapters.
  • Integration: httpx.AsyncClient + ASGITransport for routes; the socket.io test client for WebSocket.
  • Coverage ≥ 80% on service/pipeline modules.

Behavioral contracts

Invariants the brain depends on — preserve them when changing pipeline or tool-dispatch code (full list in the reference):

  • Stream TTS ahead of the LLM — synthesize each sentence as soon as a boundary (.!?\n) arrives; never wait for the full response.
  • VAD keeps a pre-speech ring buffer and is silenced while speaking/processing so the robot never hears itself.
  • Non-blocking tools are fire-and-forget (errors logged, never raised); blocking tools clear the TTS queue, speak a short ack, run, then do a follow-up LLM call.
  • Degrade, don't crash: an MCP server that fails to connect logs a warning and the app starts with reduced tools; STT/LLM/TTS errors emit an error event and return to idle.