feat(voice-bridge): STT post-correction aliases

Server-side fuzzy alias table applied to the final STT transcript
before forwarding to /voice/intent. Closes the metier-vocabulary
gap of generic STT models (U-SON, LA 440 Hz, Zacus, ...) without
needing per-request prompt biasing — which moshi-server's Rust
runtime does not expose anyway.

Loader is soft on pyyaml (degrades to no-op if missing) and on
the yaml file itself (logs + skips on parse error). Patterns are
case-insensitive with auto word-boundaries, with a (?...) escape
hatch for advanced regex.

Per-puzzle entries layer on top of globals — firmware passes
puzzle_id in the hello frame; legacy clients omit it and get
just the global polish. Loaded once at module import; restart
voice-bridge to reload.

Unit tested on 5 representative transcripts (eu son nez -> U-SON,
diapazon -> diapason, le rasus -> Zacus, le la 440 hertz -> LA 440 Hz).

game/scenarios/stt_aliases.yaml

@@ -0,0 +1,62 @@
+# STT post-correction fuzzy aliases for Kyutai (and any) STT output.
+#
+# The Kyutai 1B en_fr model does not know Zacus metier vocabulary
+# (U-SON, the 440 Hz tuning fork context, custom NPC names, etc.) and
+# the moshi-server Rust runtime does not expose any per-request
+# TextPrompt mechanism we could use to bias the LM (cf.
+# docs/voice-pipeline-roadmap.md). So we rewrite the transcript on the
+# server side before forwarding to /voice/intent.
+#
+# Loaded by tools/macstudio/voice-bridge/main.py at boot via the
+# `STT_ALIASES_PATH` env var (default: this file). Hot reload by
+# restarting voice-bridge.
+#
+# Schema:
+#   global:                # applied to every transcript
+#     - from: "regex pattern"   # case-insensitive, word boundaries auto
+#       to: "replacement"
+#   per_puzzle:            # keyed by puzzle_id (firmware passes it in hello)
+#     P1_SON:
+#       - from: "..."
+#         to: "..."
+#
+# Patterns are matched case-insensitively. Word boundaries are added
+# around the pattern unless it starts with `(?` (escape hatch for
+# bespoke regex). Replacements preserve case of the first matched
+# character.
+
+global:
+  # Professor name — the model often hears it as "the rasus" / "le rasus".
+  - from: "(?:le |la )?rasus"
+    to: "Zacus"
+  - from: "professeur ?(?:le |la )?rasus"
+    to: "Professeur Zacus"
+
+  # Common French STT polish — these are not Kyutai-specific but help
+  # everyone (e.g. whisper.cpp fallback would benefit too).
+  - from: "vue son"
+    to: "U-SON"
+  - from: "(?:eu|u) son nez"
+    to: "U-SON"
+  - from: "(?:eu|u) son(?:'s)?"
+    to: "U-SON"
+  - from: "husson(?: n'est)?"
+    to: "U-SON"
+
+per_puzzle:
+  # Puzzle U-SON — frequency / tuning fork
+  SCENE_U_SON_PROTO:
+    - from: "(?:le |la )?la (?:quatre|4) ?cent ?quarante"
+      to: "LA 440"
+    - from: "(?:le |la )?la (?:quatre|4)40 ?(?:hertz|hz)"
+      to: "LA 440 Hz"
+    - from: "diapazon"
+      to: "diapason"
+    - from: "fréquence fondamentale"
+      to: "fréquence fondamentale"
+
+  # Reserve slots for the other scenes — populate as playtest reveals
+  # actual STT failure modes per puzzle.
+  SCENE_LA_DETECTOR: []
+  SCENE_MORSE: []
+  SCENE_COFFRE: []

tools/macstudio/voice-bridge/main.py

@@ -55,6 +55,7 @@ import asyncio
 import hashlib
 import io
 import json
+import re
 import logging
 import os
 import shutil
@@ -97,6 +98,23 @@ WHISPER_URL = os.getenv("WHISPER_URL", "http://localhost:8300")
 # reason to flip this to "whispercpp" is for A/B testing or if Kyutai
 # is being upgraded.
 STT_BACKEND = os.getenv("STT_BACKEND", "kyutai").lower()
+# Post-correction fuzzy alias table for the STT output (cf.
+# game/scenarios/stt_aliases.yaml). Loaded at boot, applied to the
+# *final* transcript only (not the streaming partials, to avoid
+# stuttering UI flicker on the firmware side).
+STT_ALIASES_PATH = os.getenv(
+    "STT_ALIASES_PATH",
+    # First try sibling to main.py (Studio deploy puts the yaml next to
+    # the module), then fall back to the in-repo canonical location
+    # (dev/test from a checkout of the Zacus repo).
+    next(
+        (str(p) for p in [
+            Path(__file__).resolve().parent / "stt_aliases.yaml",
+            Path(__file__).resolve().parents[2] / "game" / "scenarios" / "stt_aliases.yaml",
+        ] if p.exists()),
+        str(Path(__file__).resolve().parent / "stt_aliases.yaml"),
+    ),
+)
 LITELLM_URL = os.getenv("LITELLM_URL", "http://localhost:4000")
 LITELLM_DEFAULT_KEY = "sk-zacus-local-dev-do-not-share"  # placeholder, log warns at boot
 LITELLM_KEY = os.environ.get("LITELLM_MASTER_KEY", LITELLM_DEFAULT_KEY)
@@ -672,6 +690,94 @@ async def health_ready() -> JSONResponse:
     return JSONResponse(content=body, status_code=code)
 
 
+# ── STT post-correction fuzzy alias table ───────────────────────────────────
+# Compiles once at module import. Format documented in
+# game/scenarios/stt_aliases.yaml. Loader silently degrades to a no-op if
+# the YAML can't be loaded — STT still works, just without metier polish.
+_STT_GLOBAL_PATTERNS: list[tuple[re.Pattern[str], str]] = []
+_STT_PUZZLE_PATTERNS: dict[str, list[tuple[re.Pattern[str], str]]] = {}
+
+
+def _compile_alias_entries(entries: Any) -> list[tuple[re.Pattern[str], str]]:
+    """Compile a list of {from, to} dicts into (Pattern, replacement) pairs.
+
+    Patterns get case-insensitive word-boundary wrapping unless they
+    already start with ``(?`` (escape hatch for advanced regex).
+    Invalid entries are logged and skipped — never raise.
+    """
+    out: list[tuple[re.Pattern[str], str]] = []
+    if not isinstance(entries, list):
+        return out
+    for raw in entries:
+        if not isinstance(raw, dict):
+            continue
+        pat = raw.get("from")
+        repl = raw.get("to")
+        if not isinstance(pat, str) or not isinstance(repl, str):
+            continue
+        wrapped = pat if pat.startswith("(?") else rf"\b(?:{pat})\b"
+        try:
+            out.append((re.compile(wrapped, re.IGNORECASE), repl))
+        except re.error as exc:
+            _jlog("stt_alias_bad_regex", pattern=pat, err=str(exc))
+    return out
+
+
+def _load_stt_aliases(path: str) -> None:
+    """Populate the module-level alias tables. Idempotent."""
+    global _STT_GLOBAL_PATTERNS, _STT_PUZZLE_PATTERNS
+    try:
+        import yaml  # local import — pyyaml is a soft dep
+    except ImportError:
+        _jlog("stt_alias_no_pyyaml")
+        return
+    p = Path(path)
+    if not p.exists():
+        _jlog("stt_alias_file_missing", path=str(p))
+        return
+    try:
+        data = yaml.safe_load(p.read_text())
+    except Exception as exc:  # noqa: BLE001
+        _jlog("stt_alias_load_err", path=str(p), err=str(exc))
+        return
+    if not isinstance(data, dict):
+        return
+    _STT_GLOBAL_PATTERNS = _compile_alias_entries(data.get("global"))
+    per_puzzle = data.get("per_puzzle") or {}
+    _STT_PUZZLE_PATTERNS = {
+        str(k): _compile_alias_entries(v)
+        for k, v in per_puzzle.items()
+        if isinstance(v, list)
+    }
+    _jlog(
+        "stt_alias_loaded",
+        path=str(p),
+        global_count=len(_STT_GLOBAL_PATTERNS),
+        puzzle_keys=list(_STT_PUZZLE_PATTERNS.keys()),
+    )
+
+
+def _apply_stt_aliases(text: str, puzzle_id: Optional[str] = None) -> str:
+    """Apply global + puzzle-scoped substitutions to a final transcript.
+
+    No-op if alias tables are empty. Puzzle-scoped patterns apply on
+    top of globals so per-puzzle entries can refine generic ones.
+    """
+    if not text:
+        return text
+    out = text
+    for pat, repl in _STT_GLOBAL_PATTERNS:
+        out = pat.sub(repl, out)
+    if puzzle_id:
+        for pat, repl in _STT_PUZZLE_PATTERNS.get(puzzle_id, ()):
+            out = pat.sub(repl, out)
+    return out
+
+
+# Load at import so /voice/ws path has the tables ready when first WS hits.
+_load_stt_aliases(STT_ALIASES_PATH)
+
+
 # ── /tts (cache → F5 primary, Piper fallback, service_down last) ─────────────
 async def _piper_fallback(text: str) -> Optional[bytes]:
     """POST text to Piper Tower :8001/synthesize. Returns WAV bytes or None.
@@ -1434,7 +1540,11 @@ async def voice_ws(ws: WebSocket) -> None:
         await ws.close(code=1003, reason="unsupported hello")
         return
     session_id = hello.get("session_id")
-    _jlog("ws_hello_ok", request_id=request_id, session_id=session_id)
+    # Optional: firmware can pass the active puzzle_id so we apply
+    # puzzle-scoped STT alias overrides on top of the global table.
+    puzzle_id = hello.get("puzzle_id") if isinstance(hello.get("puzzle_id"), str) else None
+    _jlog("ws_hello_ok", request_id=request_id, session_id=session_id,
+          puzzle_id=puzzle_id)
 
     # ── 2. Streaming loop: binary frames + text "end" ────────────────────
     buf = bytearray()
@@ -1524,6 +1634,16 @@ async def voice_ws(ws: WebSocket) -> None:
     # P1 part14: record STT call + audio duration (PCM16 16 kHz mono).
     stt_audio_s = _pcm16_seconds(len(buf), WS_EXPECTED_SR)
     await _usage_record_stt(stt_audio_s)
+    # Apply post-correction fuzzy alias table to the *final* transcript
+    # only (partials passed through verbatim above to avoid UI flicker).
+    # The result is what we forward to /voice/intent and what the
+    # firmware records — never a raw STT output with metier holes.
+    transcript_raw = transcript
+    transcript = _apply_stt_aliases(transcript_raw, puzzle_id=puzzle_id)
+    if transcript != transcript_raw:
+        _jlog("ws_stt_alias_applied", request_id=request_id,
+              before=transcript_raw[:160], after=transcript[:160],
+              puzzle_id=puzzle_id)
     await ws.send_text(json.dumps(
         {"type": "stt", "text": transcript, "final": True}
     ))

tools/macstudio/voice-bridge/requirements.txt

@@ -17,3 +17,5 @@ mlx>=0.31
 msgpack>=1.0
 websockets>=12
 scipy>=1.10
+# STT post-correction alias table loader (game/scenarios/stt_aliases.yaml).
+pyyaml>=6