feat(bot): reach host voice-server + throttle DAVE decrypt error bursts

- Log voice connection errors instead of letting them surface silently.
- Collapse bursty repeated receive-stream errors (DAVE E2EE group-transition
  decrypt failures) into one line + a suppressed-count summary, so a member
  joining/leaving no longer floods the log.

Deploy: voice-server now runs as the wsai-voice.service user unit (STT+Claude
Haiku brain+TTS on GPU); the bot container reaches it via
host.docker.internal:8787.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
EJClaw
2026-08-21 23:22:21 +09:00
parent 356d1128fa
commit 2c4c9ad82c

View File

@@ -62,6 +62,25 @@ const MIN_UTTERANCE_BYTES = Number(process.env.WSAI_MIN_UTTERANCE_BYTES || 67000
const t0 = Date.now(); const t0 = Date.now();
const log = (...a) => console.log(`[+${String(Date.now() - t0).padStart(6)}ms]`, ...a); const log = (...a) => console.log(`[+${String(Date.now() - t0).padStart(6)}ms]`, ...a);
// Throttle bursty repeated logs. DAVE (E2EE) group transitions — someone joins
// or leaves the voice channel — briefly deliver undecryptable packets, so the
// same "recv stream error" can fire many times in a second. Log the first
// occurrence of a given message immediately, then collapse repeats within a
// window into one summary line instead of flooding the log.
const _throttle = new Map(); // key -> { count, timer }
function logThrottled(key, msg, windowMs = 10_000) {
const e = _throttle.get(key);
if (e) { e.count++; return; }
log(msg);
const timer = setTimeout(() => {
const cur = _throttle.get(key);
_throttle.delete(key);
if (cur && cur.count > 0) log(`${msg} (+${cur.count} more in ${Math.round(windowMs / 1000)}s)`);
}, windowMs);
if (typeof timer.unref === 'function') timer.unref();
_throttle.set(key, { count: 0, timer });
}
// PCM s16le -> WAV container (so the Python side can ffmpeg-decode it). // PCM s16le -> WAV container (so the Python side can ffmpeg-decode it).
function wavHeader(dataLen, sampleRate = 48000, channels = 2, bits = 16) { function wavHeader(dataLen, sampleRate = 48000, channels = 2, bits = 16) {
const blockAlign = channels * bits / 8; const blockAlign = channels * bits / 8;
@@ -165,6 +184,7 @@ client.once('clientReady', async () => {
selfDeaf: false, // MUST be false to receive audio (the STT input path) selfDeaf: false, // MUST be false to receive audio (the STT input path)
selfMute: false, // false so we can also speak later (M5 TTS) selfMute: false, // false so we can also speak later (M5 TTS)
}); });
connection.on('error', (e) => log(`voice connection error: ${e.message}`));
try { try {
// The DAVE/MLS handshake here cycles signalling<->connecting several times // The DAVE/MLS handshake here cycles signalling<->connecting several times
@@ -201,7 +221,7 @@ client.once('clientReady', async () => {
// A receive-stream error (e.g. a DAVE decrypt/UDP GenericFailure on one // A receive-stream error (e.g. a DAVE decrypt/UDP GenericFailure on one
// packet) must NOT crash the process — log it and free the slot so the // packet) must NOT crash the process — log it and free the slot so the
// next utterance still works. // next utterance still works.
opusStream.on('error', (e) => { log(`recv stream error user=${userId}: ${e.message}`); active.delete(userId); }); opusStream.on('error', (e) => { logThrottled(`recv:${e.message}`, `recv stream error user=${userId}: ${e.message}`); active.delete(userId); });
opusStream.pipe(decoder); opusStream.pipe(decoder);
decoder.on('data', (d) => { decoder.on('data', (d) => {
chunks.push(d); chunks.push(d);