From 524f0dbc7345f72665bad356b70054d9bb88b538 Mon Sep 17 00:00:00 2001 From: Doxa Maps Date: Sun, 27 Sep 2026 22:08:45 +0000 Subject: [PATCH] Rudder is idle the moment you press Stop, and a new recording can start while the last words arrive On a loaded server the last words can land seconds after Stop, and the widget read "Stop" until they did. Stop now releases the mic and returns to idle at once while the session's socket stays open for the tail. Each engine session sends its events to its own reader, so a session still draining never writes into the next one, and Record again simply lets go of the old tail. --- backend/rudder-engine/main.go | 70 +++++++++----------- plugin-designs/rudder-voice/rudder-voice.vue | 26 +++++--- 2 files changed, 48 insertions(+), 48 deletions(-) diff --git a/backend/rudder-engine/main.go b/backend/rudder-engine/main.go index 4d3bb0677..6a44defb2 100644 --- a/backend/rudder-engine/main.go +++ b/backend/rudder-engine/main.go @@ -154,7 +154,6 @@ func (h *hub) send(e event) { // ─── Engine: supervises nemo-speech serve and one recording session at a time ─────────────────── type engine struct { - sink func(event) // the live browser session's own stream (RUDDER_AUDIO=browser); nil = the hub cfg config logs *ring hub *hub @@ -298,6 +297,15 @@ func (e *engine) start(lang, device string) error { return e.startWith(lang, dev // startWith starts a session. src, when set, is the audio source (a browser stream); nil spawns a local recorder. func (e *engine) startWith(lang, device string, src *recorderProc, sink func(event)) error { browser := src != nil + // A session's events go to ITS reader (a browser session's own socket), captured here, so a session + // still draining its last words after Stop never writes into the next reader's session. + emit := func(ev event) { + if sink != nil { + sink(ev) + return + } + e.hub.send(ev) + } e.mu.Lock() if e.state != "ready" { e.mu.Unlock() @@ -335,11 +343,11 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve if card := btCard(device); !browser && card != "" && os.Getenv("RUDDER_KEEP_MSBC") == "" { if p := currentProfile(card); p == "headset-head-unit-msbc" { e.logs.add("bluetooth headset on mSBC — switching to CVSD before recording (stable voice link)") - e.emit(event{"type": "recovering", "message": "Switching your Bluetooth headset to its stable voice mode…"}) + emit(event{"type": "recovering", "message": "Switching your Bluetooth headset to its stable voice mode…"}) if err := switchProfile(card, e.logs); err != nil { e.logs.add("pre-switch to CVSD failed: %v", err) } else { - e.emit(event{"type": "recovered", "message": "Headset is in its stable voice mode — go ahead and talk."}) + emit(event{"type": "recovered", "message": "Headset is in its stable voice mode — go ahead and talk."}) } } } @@ -362,7 +370,6 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve e.mu.Lock() e.recording, e.recorder, e.language, e.device = true, recName, lang, device - e.sink = sink e.stopSession = func() { cancel() recMu.Lock() @@ -382,6 +389,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve case <-time.After(wait): } ws.Close() + emit(event{"type": "recording", "recording": false}) } e.mu.Unlock() if browser { @@ -390,10 +398,10 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve e.logs.add("recording started: recorder=%s device=%q language=%s profile=%s", recName, device, lang, currentProfile(btCard(device))) go watchKernelBluetooth(ctx, e.logs) } - e.emit(event{"type": "recording", "recording": true, "recorder": recName, "language": lang, "device": device}) + emit(event{"type": "recording", "recording": true, "recorder": recName, "language": lang, "device": device}) for _, d := range e.devs.current().Devices { if (d.ID == device || (device == "" && d.Default)) && d.Muted { - e.emit(event{"type": "warning", "message": "Your microphone \"" + d.Label + "\" is muted in the system sound settings — unmute it or nothing will be heard."}) + emit(event{"type": "warning", "message": "Your microphone \"" + d.Label + "\" is muted in the system sound settings — unmute it or nothing will be heard."}) } } @@ -432,7 +440,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve } if browser { e.logs.add("browser audio stream ended: %v", err) - e.emit(event{"type": "error", "message": "The browser stopped sending microphone audio (the page was closed or lost its connection)."}) + emit(event{"type": "error", "message": "The browser stopped sending microphone audio (the page was closed or lost its connection)."}) go e.stop() return } @@ -451,7 +459,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve if why != "" { msg += " (" + r.name + ": " + why + ")" } - e.emit(event{"type": "error", "message": msg}) + emit(event{"type": "error", "message": msg}) go e.stop() return } @@ -479,7 +487,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve case recoveries == 1: // Step 1: reopen the recorder (cheap; fixes a stalled stream). e.logs.add("mic link dead (exact zeros 1.5s) — step 1: reopening the microphone") - e.emit(event{"type": "recovering", "attempt": recoveries, "message": "Microphone went silent — reconnecting it…"}) + emit(event{"type": "recovering", "attempt": recoveries, "message": "Microphone went silent — reconnecting it…"}) if next, err := spawnRecorder(device, e.logs); err == nil { recMu.Lock() old := cur @@ -494,12 +502,12 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve // "detect first", so it can't decide a dead link is fine. card := btCard(device) e.logs.add("mic still dead — step 2: forcing a new voice link: %s → headset-head-unit-cvsd", card) - e.emit(event{"type": "recovering", "attempt": recoveries, "message": "Microphone still silent — switching your Bluetooth headset to its more stable voice mode…"}) + emit(event{"type": "recovering", "attempt": recoveries, "message": "Microphone still silent — switching your Bluetooth headset to its more stable voice mode…"}) resetting.Store(true) go func() { defer resetting.Store(false) if err := switchProfile(card, e.logs); err != nil { - e.emit(event{"type": "warning", "message": "Couldn't switch the headset's voice mode: " + err.Error()}) + emit(event{"type": "warning", "message": "Couldn't switch the headset's voice mode: " + err.Error()}) return } if next, err := spawnRecorder(device, e.logs); err == nil { @@ -514,7 +522,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve case recoveries == 3 && btResetPath() != "": // Step 3: full headset reset (disconnect → restart PipeWire → reconnect). e.logs.add("mic still dead — step 3: full headset reset via RUDDER_BT_RESET") - e.emit(event{"type": "recovering", "attempt": recoveries, "message": "Still silent — fully resetting your Bluetooth headset (about 15 seconds)…"}) + emit(event{"type": "recovering", "attempt": recoveries, "message": "Still silent — fully resetting your Bluetooth headset (about 15 seconds)…"}) resetting.Store(true) go func() { defer resetting.Store(false) @@ -523,7 +531,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve return } if !ok { - e.emit(event{"type": "warning", "message": "The headset reset didn't bring the mic back — is the headset on and in range?"}) + emit(event{"type": "warning", "message": "The headset reset didn't bring the mic back — is the headset on and in range?"}) } if next, err := spawnRecorder(device, e.logs); err == nil { recMu.Lock() @@ -541,7 +549,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve if btResetPath() == "" { msg += " (No headset-reset command is configured — see RUDDER_BT_RESET.)" } - e.emit(event{"type": "warning", "message": msg}) + emit(event{"type": "warning", "message": msg}) } silentSince = time.Now() } @@ -550,7 +558,7 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve recovering.Store(false) healthySince = time.Now() e.logs.add("mic recovered after %d reconnect(s)", recoveries) - e.emit(event{"type": "recovered", "message": "Microphone is back — keep talking."}) + emit(event{"type": "recovered", "message": "Microphone is back — keep talking."}) } if recoveries > 0 && !healthySince.IsZero() && time.Since(healthySince) > 30*time.Second { recoveries, gaveUp, healthySince = 0, false, time.Time{} @@ -566,14 +574,14 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve } if time.Since(lastLevel) > 150*time.Millisecond { lastLevel = time.Now() - e.emit(event{"type": "level", "peak": float64(peak) / 32768}) + emit(event{"type": "level", "peak": float64(peak) / 32768}) } writeMu.Lock() err := ws.WriteMessage(websocket.BinaryMessage, frame) writeMu.Unlock() if err != nil { if ctx.Err() == nil { - e.emit(event{"type": "error", "message": "speech server connection lost: " + err.Error()}) + emit(event{"type": "error", "message": "speech server connection lost: " + err.Error()}) } return } @@ -598,13 +606,13 @@ func (e *engine) startWith(lang, device string, src *recorderProc, sink func(eve case strings.HasSuffix(t, "input_audio_transcription.delta"): // Deltas APPEND to the utterance in progress; the server sends many empty ones. if text != "" { - e.emit(event{"type": "partial", "text": text}) + emit(event{"type": "partial", "text": text}) } case strings.HasSuffix(t, "input_audio_transcription.completed"): - e.emit(event{"type": "final", "text": text, "at": time.Now().Format(time.RFC3339)}) + emit(event{"type": "final", "text": text, "at": time.Now().Format(time.RFC3339)}) case t == "error": e.logs.add("speech server error: %s", string(msg)) - e.emit(event{"type": "error", "message": string(msg)}) + emit(event{"type": "error", "message": string(msg)}) case t == "input_audio_buffer.committed": return default: @@ -680,26 +688,8 @@ func (e *engine) stop() { if !was || stop == nil { return } - stop() - e.logs.add("recording stopped") - e.emit(event{"type": "recording", "recording": false}) - e.mu.Lock() - e.sink = nil - e.mu.Unlock() -} - -// emit sends a session event. A browser session's events go back on that reader's own stream ONLY: on a -// shared server the hub reaches every open Rudder page, and one person's words must never appear in -// another's box. -func (e *engine) emit(ev event) { - e.mu.Lock() - s := e.sink - e.mu.Unlock() - if s != nil { - s(ev) - return - } - e.hub.send(ev) + e.logs.add("recording stopping") + stop() // sends "recording false" to the session's own reader once its last words are in } func (e *engine) shutdown() { diff --git a/plugin-designs/rudder-voice/rudder-voice.vue b/plugin-designs/rudder-voice/rudder-voice.vue index ea75359bf..6c592bcc8 100644 --- a/plugin-designs/rudder-voice/rudder-voice.vue +++ b/plugin-designs/rudder-voice/rudder-voice.vue @@ -380,6 +380,7 @@ const audioMode = ref('local') let sock = null, media = null, actx = null async function startBrowser() { + if (sock) { closeStream(); flushPartial() } // a previous session still draining its last words if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) { throw new Error('This browser cannot share its microphone here (it needs a secure https:// page).') } @@ -396,14 +397,17 @@ async function startBrowser() { const u = new URL(apiBase.value + '/stream', location.href) u.protocol = u.protocol === 'https:' ? 'wss:' : 'ws:' u.searchParams.set('language', language.value) - sock = new window.WebSocket(u) - sock.binaryType = 'arraybuffer' - sock.onmessage = (m) => onEvent(m.data) - sock.onclose = (c) => { + const ws = new window.WebSocket(u) + sock = ws + ws.binaryType = 'arraybuffer' + ws.onmessage = (m) => { if (sock === ws) onEvent(m.data) } + ws.onclose = (c) => { + if (sock !== ws) return // an older, draining session: the new one owns the mic and the box if (c.code === 4409) say(c.reason || 'Someone else is recording on this server.', 'error') sock = null releaseMic() - if (recording.value) { recording.value = false; level.value = 0; flushPartial() } + if (recording.value) { recording.value = false; level.value = 0 } + flushPartial() } node.port.onmessage = (e) => { if (sock && sock.readyState === 1) sock.send(e.data) } } @@ -413,14 +417,20 @@ function releaseMic() { media = null actx = null } -// Stop: the mic goes off at once; the socket stays open until the engine sends the last words. +// Stop: the mic goes off and the widget is idle at once (rega: it read "Stop" for 12 s+ while a loaded +// server caught up). The socket stays open so the last words still land in the box; the engine closes the +// session with "recording false". Record again meanwhile simply lets go of that tail. function stopBrowser() { releaseMic() + recording.value = false + level.value = 0 if (sock && sock.readyState === 1) sock.send(JSON.stringify({ type: 'stop' })) - else closeStream() + else { closeStream(); flushPartial() } } function closeStream() { - if (sock) { try { sock.close() } catch {} } + const ws = sock + sock = null + if (ws) { try { ws.close() } catch {} } } // Recording stopped mid-sentence: the streamed words are already in the box, so just let go of them.