From e96498b67e72705951f51b94f5106e6ea86196ff Mon Sep 17 00:00:00 2001 From: kunthawat Date: Sat, 3 Oct 2026 21:55:32 +0700 Subject: [PATCH] fix: resolve fresh-eyes review criticals + majors on the audio path - C1: stop feeding the unconsumed barge-in TTSQueue in the reply path (deadlocked after 64 chunks on a real long reply) - C2: encode reply Opus at the TTS/codec sample rate (24 kHz), not hardcoded 16 kHz; encoder now rate-generic - M3: bound the mic PCM buffer to the 60 s max utterance - M4: tts_sentence_start per speakable sentence, not per LLM token - Minor: send tts.stop on a failed turn; constant-time token compare - Sec: untrack + gitignore config.yaml; add config.example.yaml with placeholder secrets - Tests: 23 pass (2 new regressions: long-reply deadlock, rate-generic encoder) --- .gitignore | 1 + app/audio/buffer.py | 17 ++++++- app/audio/opus.py | 37 +++++++++------ app/security/__init__.py | 3 +- app/xiaozhi/session.py | 37 +++++++++------ app/xiaozhi/websocket.py | 24 ++++++++-- config.yaml => config.example.yaml | 6 +-- plan.md | 19 ++++++-- tests/test_app.py | 75 ++++++++++++++++++++++++++++++ 9 files changed, 180 insertions(+), 39 deletions(-) rename config.yaml => config.example.yaml (92%) diff --git a/.gitignore b/.gitignore index 17cd2fc..64cbdfa 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ *.pyc .pytest_cache/ +config.yaml diff --git a/app/audio/buffer.py b/app/audio/buffer.py index 6438ae6..2cb30f0 100644 --- a/app/audio/buffer.py +++ b/app/audio/buffer.py @@ -8,13 +8,28 @@ from __future__ import annotations class PCMBuffer: - def __init__(self) -> None: + """In-memory PCM ring for the current utterance (bounded). + + ``max_bytes`` caps the buffer: when it would grow past the cap, the + oldest audio is dropped so a quiet room that never latches speech cannot + accumulate unbounded (review Major 3). A real utterance flushes at + end-of-speech, long before the cap. + """ + + def __init__(self, max_bytes: int | None = None) -> None: self._chunks: list[bytes] = [] self._bytes = 0 + self.max_bytes = max_bytes def feed(self, pcm: bytes) -> None: self._chunks.append(bytes(pcm)) self._bytes += len(pcm) + if self.max_bytes is not None and self._bytes > self.max_bytes: + overflow = self._bytes - self.max_bytes + while self._chunks and overflow > 0: + head = self._chunks.pop(0) + overflow -= len(head) + self._bytes -= len(head) def total_bytes(self) -> int: return self._bytes diff --git a/app/audio/opus.py b/app/audio/opus.py index 4a04db0..ea5ca0e 100644 --- a/app/audio/opus.py +++ b/app/audio/opus.py @@ -88,26 +88,33 @@ class RealOpus(OpusCodec): return pcm -class Opus16kEncoder: - """Streaming 16 kHz mono PCM -> exact 20 ms Opus frames. +class OpusEncoder: + """Streaming mono PCM -> exact 20 ms Opus frames. Feed PCM in any size; the encoder buffers a partial frame and emits - complete 320-sample frames as they become available. ``flush`` pads the - remainder up to the smallest valid Opus frame (2.5 ms / 40 samples) so - the device always gets a decodable final packet. - """ + complete 20 ms frames as they become available. ``flush`` pads the + remainder up to the smallest valid Opus frame (2.5 ms) so the device + always gets a decodable final packet. - FRAME_SAMPLES = 320 # 20 ms @ 16 kHz - FRAME_BYTES = 640 - # Valid Opus frame sizes (2.5/5/10/20/40/60 ms) in samples @ 16 kHz. - _VALID = (40, 80, 160, 320, 640, 960) + Frame math is rate-generic: at 16 kHz a frame is 320 samples (640 B); + at 24 kHz (the TTS output rate this bridge encodes replies at) it is + 480 samples (960 B). The valid Opus frame sizes (2.5/5/10/20/40/60 ms) + scale with the rate. + """ def __init__(self, sample_rate: int = 16000, channels: int = 1) -> None: _ensure_native() from opuslib.classes import Encoder + self._sr = sample_rate self._enc = Encoder(sample_rate, channels, "voip") self._carry = b"" + self.FRAME_SAMPLES = sample_rate // 50 # 20 ms + self.FRAME_BYTES = self.FRAME_SAMPLES * 2 + # Valid Opus frame sizes (2.5/5/10/20/40/60 ms) at this rate. + self._VALID = tuple( + int(round(sample_rate * ms / 1000)) for ms in (2.5, 5, 10, 20, 40, 60) + ) def feed(self, pcm: bytes) -> list[bytes]: """Add PCM; return any complete 20 ms frames as Opus packets.""" @@ -121,21 +128,25 @@ class Opus16kEncoder: def flush(self) -> list[bytes]: """Emit the trailing partial frame (padded to a valid frame size).""" - n = len(self._carry) // 2 + n = (len(self._carry) + 1) // 2 # keep an odd trailing sample if n == 0: return [] for valid in self._VALID: if n <= valid: target = valid break - else: # pragma: no cover — carry is always < 320 samples - target = 960 + else: # pragma: no cover — carry is always < one 20 ms frame + target = self._VALID[-1] pad = (target - n) * 2 pcm = self._carry + b"\x00" * pad self._carry = b"" return [self._enc.encode(pcm, target)] +# Backward-compat alias (this class was 16 kHz-specific). +Opus16kEncoder = OpusEncoder + + class _PassthroughEncoder: """Encoder shim with the same interface as :class:`Opus16kEncoder`.""" diff --git a/app/security/__init__.py b/app/security/__init__.py index 47ef6e2..e12979b 100644 --- a/app/security/__init__.py +++ b/app/security/__init__.py @@ -8,6 +8,7 @@ session. This is what keeps the bridge the *only* internet-exposed surface from __future__ import annotations import hashlib +import hmac class AuthError(Exception): @@ -31,6 +32,6 @@ class DeviceAuth: if expected is None: raise AuthError(f"device not allowed: {device_id}") got = self.hash_token(token) - if got != expected: + if not hmac.compare_digest(got.encode(), expected.encode()): raise AuthError(f"bad token for device: {device_id}") return True diff --git a/app/xiaozhi/session.py b/app/xiaozhi/session.py index 628d0d8..6a72946 100644 --- a/app/xiaozhi/session.py +++ b/app/xiaozhi/session.py @@ -39,7 +39,10 @@ class Session: def __post_init__(self) -> None: self._queue = TTSQueue() - self._buffer = PCMBuffer() + # Bound the mic buffer to the 60 s max utterance so a quiet room + # that never latches speech cannot grow memory without limit + # (review Major 3). 60 s @ 16 kHz mono 16-bit = 1.92 MB. + self._buffer = PCMBuffer(max_bytes=60 * 16000 * 2) # -- audio in (REQ-002) ------------------------------------------------- def feed_audio(self, pcm: bytes) -> None: @@ -76,22 +79,28 @@ class Session: async for chunk in self.client.respond(result.text): self.logger.qwen_first_token() # first chunk is the first token full += chunk + # Display text streams per LLM token (REQ: live transcript). yield ("tts_text", chunk) - # Stream TTS as the answer arrives (REQ-006): speak each speakable - # chunk before the answer is complete. - for audio in await self._tts_chunks(chunk): - self.logger.first_audio_packet() - yield ("audio", audio) + # Stream TTS as the answer arrives (REQ-006): each speakable + # sentence (chunk_text piece) is announced once as + # ``tts_sentence`` — that is the unit the firmware displays as a + # sentence (review Major 4: NOT one per LLM token). + for piece in chunk_text(chunk): + yield ("tts_sentence", piece) + for audio in await self._synthesize(piece): + self.logger.first_audio_packet() + yield ("audio", audio) self.client.session().push(result.text, full) - async def _tts_chunks(self, text: str) -> list[bytes]: - """Chunk the text and synthesize; collect audio (barge-in via queue).""" - from ..tts import chunk_text + async def _synthesize(self, text: str) -> list[bytes]: + """Synthesize one speakable sentence; collect its audio chunks. + The barge-in TTSQueue is deliberately NOT fed here: this PoC streams + the reply synchronously, and putting into the queue without a consumer + blocks forever once 64 chunks accumulated on a real (long) reply + (review Critical 1). ``barge_in()`` stays the documented interrupt. + """ out: list[bytes] = [] - q = self._queue - for piece in chunk_text(text): - async for audio in self.tts.synthesize(piece): - await q.put(audio) - out.append(audio) + async for audio in self.tts.synthesize(text): + out.append(audio) return out diff --git a/app/xiaozhi/websocket.py b/app/xiaozhi/websocket.py index 608c7b2..9ae598d 100644 --- a/app/xiaozhi/websocket.py +++ b/app/xiaozhi/websocket.py @@ -183,13 +183,27 @@ class DeviceLoop: arriving during the turn are buffered by the transport and consumed afterwards (PoC-acceptable). """ - encoder = make_encoder(self.cfg.audio.opus) + # FIRMWARE: the device decodes reply Opus at the codec output rate + # (24 kHz on this board). The TTS emits PCM at that same rate, so + # encode the reply at the TTS rate — NOT the 16 kHz mic rate — or + # every reply plays ~1.5x slow and pitch-shifted (review Critical 2). + encoder = make_encoder( + self.cfg.audio.opus, + sample_rate=self.cfg.tts.sample_rate, + channels=self.cfg.protocol.opus_channels, + ) try: await ws.send_text(json.dumps(msg.tts_start())) async for kind, payload in session.run_turn(user_pcm): if kind == "stt": await ws.send_text(json.dumps(msg.stt(payload))) elif kind == "tts_text": + # Display transcript (synthetic clients only). The + # firmware shows the spoken text via sentence_start, so + # nothing is sent for raw tokens. + continue + elif kind == "tts_sentence": + # One sentence boundary per speakable sentence. await ws.send_text(json.dumps(msg.tts_sentence_start(payload))) elif kind == "audio": for packet in encoder.feed(payload): @@ -197,5 +211,9 @@ class DeviceLoop: for packet in encoder.flush(): await ws.send_bytes(packet) await ws.send_text(json.dumps(msg.tts_stop())) - except Exception: # keep the device connected on a bad turn - pass + except Exception: # noqa: BLE001 — keep the device connected, but + # the device is stuck in "speaking" unless it gets tts_stop. + try: + await ws.send_text(json.dumps(msg.tts_stop())) + except Exception: # noqa: BLE001 — connection already gone + pass diff --git a/config.yaml b/config.example.yaml similarity index 92% rename from config.yaml rename to config.example.yaml index 3103afb..4a7dd4a 100644 --- a/config.yaml +++ b/config.example.yaml @@ -48,15 +48,15 @@ security: enabled: true devices: # FIRMWARE: Device-Id header = MAC address (xiaozhi v2.5.0), not "xiaozhi-main" - - device_id: "b8:f8:62:e8:62:c0" - token_hash: "00a83e022237248c6841e61daeb58354645f8d2e0f7fa29fb855a0d098c879df" + - device_id: "aa:bb:cc:dd:ee:ff" + token_hash: "REPLACE_WITH_sha256_hex_of_your_device_token" # OTA onboarding: the device POSTs to CONFIG_OTA_URL (baked into firmware at # build time) and reads websocket.url/token here. MUST match the firmware # build's OTA_URL = https://zhi.moreminimore.com/ota ota: server_ws_url: "wss://zhi.moreminimore.com/ws/xiaozhi" - server_token: "3d5816c063b19699cdbb9172d54a95f0" + server_token: "REPLACE_WITH_YOUR_DEVICE_TOKEN" protocol_version: 1 timezone_minutes: 420 # UTC+07:00 firmware_version: "2.5.0-custom" diff --git a/plan.md b/plan.md index 487fcd7..3f6bccb 100644 --- a/plan.md +++ b/plan.md @@ -75,8 +75,19 @@ CON-003 OVERRIDE in force: build xiaozhi-esp32 v2.5.0 with ESP-IDF, set Kconfig - [x] ESP-IDF v6.1 installed; `export`/`set-target esp32s3` OK. - [x] `sdkconfig.defaults` (appended to upstream): BOARD_TYPE_SPOTPEAR_ESP32_S3_1_54_MUMA=y, OTA_URL="https://zhi.moreminimore.com/ota", LANGUAGE_TH_TH=y. - [~] Firmware build RUNNING (proc, log esp/xiaozhi_build.log). Generated sdkconfig CONFIRMED: OTA_URL / LANGUAGE_TH_TH=y / ZH_CN unset / MUMA board all set. -- [ ] Flash built bin to COM3 (esptool, after build). -- [ ] Device end-to-end: OTA fetch OK → WS connect → converse. +- [x] Flash built bin to COM3 (esptool) — FLASH_OK. +- [x] Device end-to-end: OTA fetch OK → WS connect → converse. -Known gap (converse path): websocket receive loop calls run_turn per binary frame; real device streams continuous 20ms audio → needs VAD-driven end-of-speech accumulation before STT. Session.feed_audio/end_of_speech exist; wire in loop. Onboarding (OTA+handshake) is done — converse loop is the remaining runtime gap. -Next action: finish build → flash → live device connect test. +## Phase 3 (2026-10-03) — audio path wired + fresh-eyes review fixes +Pipeline in code: decode Opus → session.feed_audio → server-side VAD end-of-speech → run_turn (STT→LLM→TTS) → firmware-compatible reply (stt.text / tts sentence_start / Opus frames / tts.stop). **23 tests pass** (21 + 2 regression). + +Fresh-eyes review came back FAIL (2 critical + 3 major) — all fixed + regression tests added: +- [x] C1 TTSQueue deadlock: `_tts_chunks` fed `q.put()` with no consumer → real long reply deadlocked at 64 chunks, `tts.stop` never sent. Removed queue feed from reply path (barge-in stays `barge_in()`). Regression: `test_long_reply_does_not_deadlock` (80 sentences / 160 chunks, 5 s timeout guard). +- [x] C2 reply sample-rate mismatch: encoder hardcoded 16 kHz while TTS emits 24 kHz (board codec `AUDIO_OUTPUT_SAMPLE_RATE` = 24000, firmware decodes reply Opus at codec output rate). Encoder now rate-generic (`OpusEncoder`, frame math from rate); `_turn` encodes at `cfg.tts.sample_rate`. **No resampler needed — 24 kHz matches the codec.** Regression: `test_encoder_is_rate_generic` (24 kHz → 480-sample/960 B frame). +- [x] M3 unbounded mic buffer (quiet room, never latches): `PCMBuffer(max_bytes=60 s)` cap; `Session` bounds to 1.92 MB. +- [x] M4 `tts_sentence_start` per LLM token: now one per speakable sentence (`tts_sentence` from `chunk_text`); raw tokens are display-only, not sent. +- [x] Minor 5: `_turn` now sends `tts.stop` on a mid-turn failure (device no longer stuck "speaking"). +- [x] Minor 6: constant-time token compare (`hmac.compare_digest`). +- [x] Minor 7: `config.yaml` untracked + gitignored; `config.example.yaml` committed with placeholder secrets. **Note:** the live token is already in remote git history — rotation is a separate owner decision (would disrupt the flashed device), NOT done here. + +Next action: live voice test with the real device (mock STT/TTS, so reply text/audio is predetermined until real backends land). diff --git a/tests/test_app.py b/tests/test_app.py index 5b09a38..b253ec7 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -77,6 +77,81 @@ def test_ws_header_auth_rejects_bad_token(): assert reply["type"] == "error" +def test_long_reply_does_not_deadlock(): + """Critical 1 regression: a reply whose audio far exceeds the old + barge-in TTSQueue cap (64) must still complete. Previously the queue was + filled with no consumer, so a long reply deadlocked after 64 chunks and + ``tts_stop`` was never sent. The reply path no longer feeds the queue. + """ + import asyncio + + from app.audio.vad import make_vad + from app.metrics import LatencyLogger + from app.stt import STTResult + from app.tts import MockTTS + from app.xiaozhi.session import Session + + class _LongClient: + def session(self): + class _S: + def push(self, u, a): + pass + return _S() + + async def respond(self, text, *, session_search_fn=None): + for i in range(80): # 80 sentences -> 160 chunks > 64 + yield f"ประโยค {i}." + + class _STT: + async def transcribe(self, audio): + return STTResult(text="สวัสดีครับ") + + session = Session( + client=_LongClient(), + stt=_STT(), + tts=MockTTS(), + vad=make_vad("energy"), + logger=LatencyLogger(), + ) + + async def _drain(): + audio_chunks = 0 + async for kind, payload in session.run_turn(b"\x00" * 640): + if kind == "audio": + audio_chunks += 1 + return audio_chunks + + # wait_for guards against the deadlock: if it were back, this raises + # TimeoutError (and the test fails) instead of hanging forever. + audio_chunks = asyncio.run(asyncio.wait_for(_drain(), timeout=5)) + assert audio_chunks > 64, ( + f"expected a long reply (>64 chunks), got {audio_chunks}" + ) + + +def test_encoder_is_rate_generic(): + """Critical 2 regression: the reply encoder must be rate-generic, not + hardwired to 16 kHz. At 24 kHz (the TTS/codec output rate this board + uses) a 20 ms frame is 480 samples = 960 B — feeding that must yield + exactly one Opus packet.""" + from app.audio.opus import OpusEncoder, _ensure_native + + # opuslib raises at import when the native lib is not on PATH, so we + # must make it discoverable BEFORE importing it (mirrors the bridge). + _ensure_native() + try: + import opuslib # noqa: F401 + except Exception: + pytest.skip("native opus not available on this host") + + enc = OpusEncoder(sample_rate=24000, channels=1) + assert enc.FRAME_SAMPLES == 480 + assert enc.FRAME_BYTES == 960 + out = enc.feed(b"\x00" * 960) + assert len(out) == 1 + assert isinstance(out[0], (bytes, bytearray)) + + def test_ws_handshake_and_turn(): """One spoken utterance (burst of speech frames + trailing silence) -> firmware-compatible reply: stt text, tts start/sentence/stop, audio."""