Rudder keeps streaming on a thin client: reconnect a broken speech stream, say when it falls behind

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.
2026-09-28 15:31:19 +00:00 · 2026-09-28 15:31:19 +00:00 · 2bff270734
commit 2bff270734
parent 3c3bde71e4
4 changed files with 160 additions and 61 deletions
--- 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
 }

--- 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)
+}
--- 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) {}
--- 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")
 	}