doxa-life/doxa-maps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Rudder is idle the moment you press Stop, and a new recording can start while the last words arrive

Browse source 
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.
This commit is contained in:
Doxa Maps 2026-09-27 22:08:45 +00:00
parent ec47023575
commit 1303d6a41a
2 changed files with 48 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
backend/rudder-engine/main.go
View file

@ -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() {

26
plugin-designs/rudder-voice/rudder-voice.vue
View file

@ -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.