From 2bff270734a993e8a1d1889cde121af032f3c87a Mon Sep 17 00:00:00 2001 From: Doxa Maps Date: Mon, 28 Sep 2026 15:31:19 +0000 Subject: [PATCH] Rudder keeps streaming on a thin client: reconnect a broken speech stream, say when it falls behind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Optimising the default for an average laptop, not more CPU (the speech model is hard-wired to 4 threads in NeMo-Speech.cpp, session.cpp:911 — already a thin-client budget). - RECONNECT: a live recording ended with "speech server connection lost: … broken pipe" when the stream to the speech server broke. Now the engine opens a new stream and keeps going; the words still in flight on the old one are finalised so the box keeps them; up to 5 attempts, then a plain message. Verified against a fake speech server that drops the first stream after 3 s: 6 words finalised, reconnected in 2 ms, 14 more words streamed, clean Stop, 0 data races (go -race). - FALLING BEHIND: when the mic clearly carries speech but no words have come back for 10 s (the server is starved of CPU — the stream used to freeze with no message), say so once; it clears when words resume. Verified with a fake server that returns no words: warned after 10 s. - The robot voice (Piper) runs at nice 10 on Linux, so its ~1 s burst per sentence never makes the words you are speaking fall behind. --- backend/rudder-engine/main.go | 204 +++++++++++++++++------- backend/rudder-engine/procattr_linux.go | 7 + backend/rudder-engine/procattr_other.go | 3 + backend/rudder-engine/tts.go | 7 +- 4 files changed, 160 insertions(+), 61 deletions(-) diff --git a/backend/rudder-engine/main.go b/backend/rudder-engine/main.go index c0d319823..b4da4b9d5 100644 --- a/backend/rudder-engine/main.go +++ b/backend/rudder-engine/main.go @@ -346,17 +346,25 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve } e.mu.Unlock() - ws, _, err := websocket.DefaultDialer.Dial("ws://127.0.0.1:"+e.cfg.NemoPort+"/v1/audio/transcriptions/realtime", - http.Header{"Authorization": {"Bearer " + e.nemoKey}}) - if err != nil { - return fmt.Errorf("connect to speech server: %w", err) + // dial opens a transcription stream on the speech server; used again to RECONNECT mid-recording. + dial := func() (*websocket.Conn, error) { + c, _, err := websocket.DefaultDialer.Dial("ws://127.0.0.1:"+e.cfg.NemoPort+"/v1/audio/transcriptions/realtime", + http.Header{"Authorization": {"Bearer " + e.nemoKey}}) + if err != nil { + return nil, fmt.Errorf("connect to speech server: %w", err) + } + upd, _ := json.Marshal(event{"type": "session.update", "session": event{ + "sample_rate": 16000, "language": lang, "automatic_punctuation": true, "endpointing_ms": e.cfg.EndpointMs, + }}) + if err := c.WriteMessage(websocket.TextMessage, upd); err != nil { + c.Close() + return nil, fmt.Errorf("session.update: %w", err) + } + return c, nil } - upd, _ := json.Marshal(event{"type": "session.update", "session": event{ - "sample_rate": 16000, "language": lang, "automatic_punctuation": true, "endpointing_ms": e.cfg.EndpointMs, - }}) - if err := ws.WriteMessage(websocket.TextMessage, upd); err != nil { - ws.Close() - return fmt.Errorf("session.update: %w", err) + ws, err := dial() + if err != nil { + return err } // Bluetooth headsets start on CVSD, not mSBC. Measured 2026-09-27 on a Bluetooth headset: on mSBC the @@ -388,8 +396,10 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve cur := first // the live recorder; replaced when a dead mic link is recovered ctx, cancel := context.WithCancel(context.Background()) - var writeMu sync.Mutex - done := make(chan struct{}) + var writeMu sync.Mutex // guards ws and done, which a reconnect replaces + var done chan struct{} + var lastWords atomic.Int64 // unix ms of the last text from the speech server (falling-behind detector) + lastWords.Store(time.Now().UnixMilli()) e.mu.Lock() e.recording, e.recorder, e.language, e.device = true, recName, lang, device @@ -407,6 +417,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve _ = ws.WriteMessage(websocket.BinaryMessage, silence) } _ = ws.WriteMessage(websocket.TextMessage, []byte(`{"type":"input_audio_buffer.commit"}`)) + curWS, curDone := ws, done writeMu.Unlock() // Let the final words arrive. A shared server's CPU can run behind real time (measured: a loaded // 8-core box finished an 11 s clip ~15 s after it ended), so a browser session waits longer. @@ -415,10 +426,10 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve wait = 30 * time.Second } select { - case <-done: + case <-curDone: case <-time.After(wait): } - ws.Close() + curWS.Close() emit(event{"type": "recording", "recording": false}) } e.mu.Unlock() @@ -435,6 +446,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve } } + var reconnect func(why string) // assigned below, with the reader it restarts // Audio pump: 100 ms frames → speech server, plus a level meter and dead-mic RECOVERY. go func() { frame := make([]byte, 3200) @@ -443,6 +455,8 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve var recovering atomic.Bool var resetting atomic.Bool tracePeak, lastTrace := 0, time.Now() + var lastLoud time.Time + behindWarned := false var healthySince time.Time // a mic that stays healthy 30 s earns back its 3 reconnects gaveUp := false lastLevel := time.Time{} @@ -612,62 +626,132 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve lastLevel = time.Now() emit(event{"type": "level", "peak": float64(peak) / 32768}) } + // ★ FALLING BEHIND: the mic clearly carries speech but no words have come back for 10 s — the speech + // server is starved of CPU (measured 2026-09-28: another heavy process maxed the CPU and the stream + // froze with no message). Say so once; it clears as soon as words arrive again. + if peak > 2500 { + lastLoud = time.Now() + } + if !behindWarned && time.Since(lastLoud) < 2*time.Second && time.Now().UnixMilli()-lastWords.Load() > 10000 { + behindWarned = true + e.logs.add("falling behind: speech in, no words for 10 s") + emit(event{"type": "warning", "message": "The speech engine is falling behind — another program is using a lot of CPU. Your words will catch up; closing heavy programs helps."}) + } else if behindWarned && time.Now().UnixMilli()-lastWords.Load() < 2000 { + behindWarned = false + emit(event{"type": "recovered", "message": "Caught up."}) + } writeMu.Lock() err := ws.WriteMessage(websocket.BinaryMessage, frame) writeMu.Unlock() if err != nil { - if ctx.Err() == nil { - emit(event{"type": "error", "message": "speech server connection lost: " + err.Error()}) + if ctx.Err() != nil { + return } - return + reconnect("send: " + err.Error()) // keep streaming; frames during the reconnect are dropped + continue } } }() - // Result reader: map the server's events onto the plugin's small vocabulary. - go func() { - defer close(done) - committed := false - for { - _, msg, err := ws.ReadMessage() - if err != nil { - return - } - var m map[string]any - if json.Unmarshal(msg, &m) != nil { - continue - } - t, _ := m["type"].(string) - text := firstString(m, "transcript", "delta", "text") - switch { - case strings.HasSuffix(t, "input_audio_transcription.delta"): - if committed { - _ = ws.SetReadDeadline(time.Now().Add(6 * time.Second)) - } - // Deltas APPEND to the utterance in progress; the server sends many empty ones. - if text != "" { - emit(event{"type": "partial", "text": text}) - } - case strings.HasSuffix(t, "input_audio_transcription.completed"): - emit(event{"type": "final", "text": text, "at": time.Now().Format(time.RFC3339)}) - if committed { - return // the last utterance after Stop is in - } - case t == "error": - e.logs.add("speech server error: %s", string(msg)) - emit(event{"type": "error", "message": string(msg)}) - case t == "input_audio_buffer.committed": - // The commit acknowledges Stop; the words for the audio before it can still follow (rega: a - // Stop mid-word lost "what your country…"). Keep reading until the final text or 6 s of quiet (a loaded server can take seconds per backlog chunk). - committed = true - e.logs.add("speech server committed the stream; reading its last words") - _ = ws.SetReadDeadline(time.Now().Add(6 * time.Second)) - continue - default: - e.logs.add("speech event: %s", t) - } + // Result reader: map the server's events onto the plugin's small vocabulary. One reader per stream; + // a RECONNECT starts a new one on the new stream. + var reconnecting atomic.Bool + var startReader func(ws *websocket.Conn) + // ★ RECONNECT (measured 2026-09-28: "speech server connection lost: … broken pipe" ended a live + // recording). When the stream to the speech server breaks mid-recording, open a new one and keep going. + // The words that were still in flight on the old stream are finalised, so the box keeps them. + reconnect = func(why string) { + if ctx.Err() != nil || !reconnecting.CompareAndSwap(false, true) { + return } - }() + e.logs.add("speech stream broke (%s) — reconnecting", why) + emit(event{"type": "recovering", "message": "Reconnecting to the speech engine…"}) + go func() { + defer reconnecting.Store(false) + for attempt := 1; attempt <= 5 && ctx.Err() == nil; attempt++ { + nws, err := dial() + if err == nil { + writeMu.Lock() + old := ws + ws = nws + startReader(nws) + writeMu.Unlock() + old.Close() + lastWords.Store(time.Now().UnixMilli()) + e.logs.add("speech stream reconnected (attempt %d)", attempt) + emit(event{"type": "recovered", "message": "Reconnected — keep talking."}) + return + } + e.logs.add("reconnect attempt %d failed: %v", attempt, err) + time.Sleep(time.Duration(attempt) * time.Second) + } + if ctx.Err() == nil { + emit(event{"type": "error", "message": "The speech engine stopped answering. Press Record to try again."}) + go e.stop() + } + }() + } + startReader = func(ws *websocket.Conn) { + d := make(chan struct{}) + done = d + go func() { + defer close(d) + committed := false + var inFlight strings.Builder // this stream's words not yet finalised + for { + _, msg, err := ws.ReadMessage() + if err != nil { + if !committed && ctx.Err() == nil { + if inFlight.Len() > 0 { + emit(event{"type": "final", "text": inFlight.String(), "at": time.Now().Format(time.RFC3339)}) + } + reconnect("read: " + err.Error()) + } + return + } + var m map[string]any + if json.Unmarshal(msg, &m) != nil { + continue + } + t, _ := m["type"].(string) + text := firstString(m, "transcript", "delta", "text") + switch { + case strings.HasSuffix(t, "input_audio_transcription.delta"): + if committed { + _ = ws.SetReadDeadline(time.Now().Add(6 * time.Second)) + } + // Deltas APPEND to the utterance in progress; the server sends many empty ones. + if text != "" { + lastWords.Store(time.Now().UnixMilli()) + inFlight.WriteString(text) + emit(event{"type": "partial", "text": text}) + } + case strings.HasSuffix(t, "input_audio_transcription.completed"): + lastWords.Store(time.Now().UnixMilli()) + inFlight.Reset() + emit(event{"type": "final", "text": text, "at": time.Now().Format(time.RFC3339)}) + if committed { + return // the last utterance after Stop is in + } + case t == "error": + e.logs.add("speech server error: %s", string(msg)) + emit(event{"type": "error", "message": string(msg)}) + case t == "input_audio_buffer.committed": + // The commit acknowledges Stop; the words for the audio before it can still follow (rega: a + // Stop mid-word lost "what your country…"). Keep reading until the final text or 6 s of quiet (a loaded server can take seconds per backlog chunk). + committed = true + e.logs.add("speech server committed the stream; reading its last words") + _ = ws.SetReadDeadline(time.Now().Add(6 * time.Second)) + continue + default: + e.logs.add("speech event: %s", t) + } + } + }() + } + writeMu.Lock() + startReader(ws) + writeMu.Unlock() return nil } diff --git a/backend/rudder-engine/procattr_linux.go b/backend/rudder-engine/procattr_linux.go index c0d6b0121..3607c9388 100644 --- a/backend/rudder-engine/procattr_linux.go +++ b/backend/rudder-engine/procattr_linux.go @@ -10,3 +10,10 @@ import "syscall" func childAttr() *syscall.SysProcAttr { return &syscall.SysProcAttr{Pdeathsig: syscall.SIGKILL} } + +// yieldCPU lowers a background helper's CPU priority (nice 10), so on a thin client the live transcription +// always wins the CPU over it. Used for the robot voice: its ~1 s burst per sentence must never make the +// words you are speaking fall behind. +func yieldCPU(pid int) { + _ = syscall.Setpriority(syscall.PRIO_PROCESS, pid, 10) +} diff --git a/backend/rudder-engine/procattr_other.go b/backend/rudder-engine/procattr_other.go index 2c17ee968..6c57da041 100644 --- a/backend/rudder-engine/procattr_other.go +++ b/backend/rudder-engine/procattr_other.go @@ -10,3 +10,6 @@ import "syscall" func childAttr() *syscall.SysProcAttr { return nil } + +// yieldCPU: best effort only on Linux (see procattr_linux.go); elsewhere a no-op. +func yieldCPU(pid int) {} diff --git a/backend/rudder-engine/tts.go b/backend/rudder-engine/tts.go index 3b593512a..04fa12ce8 100644 --- a/backend/rudder-engine/tts.go +++ b/backend/rudder-engine/tts.go @@ -139,7 +139,12 @@ func (t *ttsEngine) synthesize(ctx context.Context, text, voiceID string, speake var pcm, stderr bytes.Buffer cmd.Stdout, cmd.Stderr = &pcm, &stderr start := time.Now() - if err := cmd.Run(); err != nil { + err = cmd.Start() + if err == nil { + yieldCPU(cmd.Process.Pid) // speech-to-text keeps priority over the robot voice (thin clients) + err = cmd.Wait() + } + if err != nil { t.logs.add("robot voice failed after %s: %v", time.Since(start).Round(time.Millisecond), err) return nil, errors.New("the voice engine failed — see the engine log") }