From d3fcc1dfdb945fbbb8f844649cf5b58ac73b3a41 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 14:19:35 +0400 Subject: [PATCH] mavwaked: throw away the round-trip backlog before it becomes a turn Nothing reads the microphone while Send is in flight, so the audio piles up in arecord's pipe and arrives in a burst the moment dispatch returns. A round-trip is p50 2.7s through the LLM router, which is about 90 frames of room, of him finishing his sentence, of the television. The old code reset the VAD on the reply path only, and for a reason that was not true: the comment said the VAD had been accumulating during the round-trip, when its state is exactly what Feed left it as. The two paths with no reset are the ones that mattered, because neither starts playback and so neither is covered by the half-duplex gate. A text-only turn fed the whole backlog into the VAD, and a Send error did the same on every failed turn, so a dead socket drove a retry loop off backlog alone. The backlog was scored for barge-in too. Five frames delivered in microseconds cut her off with audio recorded before she started speaking, which is the opposite of what the five-frame guard is for. Both are fixed by the same mechanism: measure the wall time the round-trip took, convert it to frames, and discard that many before anything looks at them. Barge-in also threw away the 150ms that proved he was talking. The VAD started from the next frame, so the first word of a short interruption was clipped before whisper saw it. Those frames are kept in a small ring and replayed after the reset. A stuck aplay was worse than before this feature existed. Playing() gates all capture, so a wedged child made her deaf rather than silent, for the full 30s ceiling inherited from the fire-and-forget version. The mute window is bounded by the reply's own duration plus a margin now. Three smaller ones. "-barge-in -barge-in-rms 0" logged "barge-in on" and then did nothing. The sent counter incremented before the error check, so failed round-trips counted as shipped. And the threshold the operator has to guess is now reported: mavwaked logs the mean energy of the frames it suppressed while speaking, so he can set it from data. Found in review of #76. --- cmd/mavwaked/main.go | 19 ++++- cmd/mavwaked/playback.go | 17 +++- cmd/mavwaked/session.go | 126 ++++++++++++++++++++++++++++-- cmd/mavwaked/session_test.go | 145 +++++++++++++++++++++++++++++++++++ 4 files changed, 293 insertions(+), 14 deletions(-) diff --git a/cmd/mavwaked/main.go b/cmd/mavwaked/main.go index 60968f2..65e4a55 100644 --- a/cmd/mavwaked/main.go +++ b/cmd/mavwaked/main.go @@ -14,9 +14,12 @@ // // While a reply is playing the capture side is muted (half-duplex): without // it, Maven's own voice comes back in through the mic and she answers -// herself. -barge-in punches one hole in that gate — sustained energy well -// above the speaker's leak level cuts playback so he can talk over her. It is -// off by default because the threshold is room-specific; see playback.go. +// herself. -barge-in punches one hole in that gate — sustained energy above +// -barge-in-rms cuts playback so he can talk over her. It is off by default +// because the threshold is room-specific; see playback.go. The threshold is a +// raw frame RMS and has no reference to what the speaker actually leaks, so +// the daemon logs the mean energy of the frames it suppressed while speaking. +// Set -barge-in-rms from those numbers rather than by guessing. // // usage: // mavwaked # default ALSA device, 127.0.0.1:9100 @@ -130,7 +133,15 @@ func run(args []string) error { var barge bargeInConfig if *bargeIn { barge = bargeInConfig{RMS: float64(*bargeRMS) / 10000.0, Frames: *bargeFrames} - log.Printf("mavwaked: barge-in on (rms %.4f x %d frames)", barge.RMS, barge.Frames) + if barge.Enabled() { + log.Printf("mavwaked: barge-in on (rms %.4f x %d frames)", barge.RMS, barge.Frames) + } else { + // The log used to say "barge-in on (rms 0.0000 x 5)" here and then + // nothing happened, because Enabled needs a positive threshold. + log.Printf("mavwaked: -barge-in was passed but rms %.4f x %d frames disables it; "+ + "both must be above zero, so barge-in is OFF", + barge.RMS, barge.Frames) + } } sess := newSession(vad, newAplayPlayer(), &voiceSender{vc: vc}, *lang, barge) diff --git a/cmd/mavwaked/playback.go b/cmd/mavwaked/playback.go index 7bc5e6d..01fd69d 100644 --- a/cmd/mavwaked/playback.go +++ b/cmd/mavwaked/playback.go @@ -33,6 +33,11 @@ import ( "github.com/kami/maven/internal/audio" ) +// playbackMargin is the slack over the reply's own duration before a stuck +// aplay is killed. Enough for ALSA to open the device and drain its buffer, +// short enough that a busy device does not cost her a turn. +const playbackMargin = 2 * time.Second + // player plays one reply at a time and can be cut off mid-utterance. type player interface { // Play starts playback of a, replacing anything already playing, and @@ -87,6 +92,13 @@ func (p *aplayPlayer) Play(a audio.Audio) { p.playing = true p.mu.Unlock() + // Bound the mute window by the reply itself. Playing() gates all capture + // now, so a wedged aplay does not merely go silent, it makes her deaf for + // as long as the flag is set. The old ceiling was a flat 30s inherited + // from the fire-and-forget version, where it only bounded a leaked + // goroutine. A reply cannot legitimately take longer than it lasts. + limit := time.Duration(a.Duration()*float64(time.Second)) + playbackMargin + go func() { if _, err := stdin.Write(a.Bytes); err != nil { // Broken pipe is the expected outcome of Stop(). @@ -101,8 +113,9 @@ func (p *aplayPlayer) Play(a audio.Audio) { if err != nil { log.Printf("mavwaked: aplay: %v", err) } - case <-time.After(30 * time.Second): - log.Printf("mavwaked: aplay timeout, killing") + case <-time.After(limit): + log.Printf("mavwaked: aplay did not finish %.1fs of audio within %s, killing (capture was muted the whole time)", + a.Duration(), limit) if pr := cmd.Process; pr != nil { _ = pr.Kill() } diff --git a/cmd/mavwaked/session.go b/cmd/mavwaked/session.go index 21e7a68..24de887 100644 --- a/cmd/mavwaked/session.go +++ b/cmd/mavwaked/session.go @@ -7,6 +7,7 @@ package main import ( "context" "log" + "time" "github.com/kami/maven/internal/audio" ) @@ -42,6 +43,20 @@ type session struct { lang string barge bargeInConfig + // now is the clock, swapped in tests. The round-trip backlog is measured + // in wall time, because that is the only thing that says how much room + // went into the pipe while the daemon was thinking. + now func() time.Time + + // discard is how many buffered frames still have to be thrown away + // before capture means anything again. See dispatch. + discard int + + // recent holds the last few frames seen during playback, so the ones + // that proved he was interrupting can be replayed into the VAD after the + // barge-in reset instead of being clipped off the front of his sentence. + recent [][]byte + // loudFrames counts consecutive over-threshold frames seen while she is // speaking. Reset whenever a frame falls back under the threshold, and // whenever playback ends. @@ -49,14 +64,28 @@ type session struct { // counters, read by tests and logged on the way out. suppressed int // frames dropped because she was speaking + dropped int // frames dropped as round-trip backlog bargeIns int // times playback was cut because he spoke over her sent int // utterances shipped to the daemon + + // loudSum and loudSeen accumulate the energy of suppressed frames, so + // the operator can read what the room actually measures and set + // -barge-in-rms from data instead of guessing. + loudSum float64 + loudSeen int } func newSession(vad *VAD, p player, s utteranceSender, lang string, barge bargeInConfig) *session { - return &session{vad: vad, player: p, sender: s, lang: lang, barge: barge} + return &session{vad: vad, player: p, sender: s, lang: lang, barge: barge, now: time.Now} } +// frameDuration is the wall time one captured frame represents. +const frameDuration = defaultFrameMs * time.Millisecond + +// suppressLogEvery — how many suppressed frames between energy reports. 200 +// frames is six seconds of her talking, so this is roughly one line per reply. +const suppressLogEvery = 200 + // feed processes one 30ms PCM frame. // // While the player is running the capture side is muted: the VAD is not fed @@ -65,26 +94,52 @@ func newSession(vad *VAD, p player, s utteranceSender, lang string, barge bargeI // energy well above the speaker's leak level cuts playback, and capture // resumes on the very next frame with a clean VAD. func (s *session) feed(ctx context.Context, frame []byte) error { + // Backlog first, before anything looks at this frame. These are frames + // the microphone captured while the round-trip blocked; they arrive in a + // burst at pipe speed and they are not a command, not an answer and not + // an interruption. + if s.discard > 0 { + s.discard-- + s.dropped++ + return nil + } + if s.player.Playing() { s.suppressed++ + rms := frameRMS(PCMToI16(frame)) + s.loudSum += rms + s.loudSeen++ + if s.loudSeen >= suppressLogEvery { + // The doc comment asks for energy "well above the speaker's leak + // level" and never says what that is. This is what it is. + log.Printf("mavwaked: suppressed %d frames while speaking, mean rms %.4f (barge-in threshold %.4f)", + s.loudSeen, s.loudSum/float64(s.loudSeen), s.barge.RMS) + s.loudSum, s.loudSeen = 0, 0 + } if !s.barge.Enabled() { return nil } - if frameRMS(PCMToI16(frame)) < s.barge.RMS { + if rms < s.barge.RMS { s.loudFrames = 0 + s.recent = s.recent[:0] return nil } s.loudFrames++ + s.keepRecent(frame) if s.loudFrames < s.barge.Frames { return nil } // He is talking over her. Cut her off, drop the VAD state that - // accumulated from the echo, and start listening for real. + // accumulated from the echo, and start listening for real — starting + // with the frames that proved he was talking. Those used to be + // thrown away, which clipped the first 150ms off his interruption, + // and on a short one that is the whole first word. s.player.Stop() s.bargeIns++ s.loudFrames = 0 s.vad.Reset() log.Printf("mavwaked: barge-in — stopped playback") + s.replayRecent() return nil } @@ -102,22 +157,77 @@ func (s *session) feed(ctx context.Context, frame []byte) error { return s.dispatch(ctx, utt) } +// keepRecent stores a copy of one barge-in trigger frame, keeping at most +// barge.Frames of them. +func (s *session) keepRecent(frame []byte) { + if len(s.recent) >= s.barge.Frames { + copy(s.recent, s.recent[1:]) + s.recent = s.recent[:len(s.recent)-1] + } + s.recent = append(s.recent, append([]byte(nil), frame...)) +} + +// replayRecent feeds the trigger frames back into the freshly reset VAD, so +// his interruption starts where he started it. +// +// Feed cannot complete an utterance here: closing one needs silenceMs of +// trailing quiet and these frames are all above the barge-in threshold, which +// is far above the VAD floor. Any utterance it did return would be a fragment +// of a sentence he is still speaking, so it is not dispatched. +func (s *session) replayRecent() { + for _, f := range s.recent { + s.vad.Feed(PCMToI16(f)) + } + s.recent = s.recent[:0] +} + // dispatch ships a complete utterance and plays whatever comes back. +// +// Every return path here has to deal with the backlog. Nothing reads the +// microphone while Send is in flight, so the audio piles up in arecord's pipe +// and the kernel buffer, and it arrives in a burst the moment this returns. A +// round-trip is p50 2.7s through the LLM router, which is around 90 frames of +// room, of him finishing his sentence, of the television. +// +// This used to reset the VAD on the reply path only, and for the wrong reason: +// the comment said the VAD had been accumulating during the round-trip, when +// in fact its state is exactly what Feed left it as. The two paths that had no +// reset are the ones that mattered, because neither of them starts playback +// and so neither is covered by the half-duplex gate. A text-only turn fed the +// whole backlog straight into the VAD, and a Send error did the same on every +// failed turn, so a dead socket drove a retry loop off nothing but backlog. func (s *session) dispatch(ctx context.Context, utt audio.Audio) error { log.Printf("mavwaked: utterance complete (%.2fs, %d bytes), sending...", utt.Duration(), len(utt.Bytes)) + start := s.now() reply, err := s.sender.Send(ctx, utt, s.lang) - s.sent++ + defer s.dropBacklog(start) if err != nil { return err } + // Count what was shipped, not what was attempted. This used to run + // before the error check, so failed round-trips counted as sent. + s.sent++ if len(reply.Bytes) == 0 { log.Printf("mavwaked: empty reply audio (text only)") return nil } - // The VAD has been accumulating from the buffered mic stream while the - // round-trip blocked. None of it is a command — reset before the - // speaker opens, so the first post-reply frame starts clean. - s.vad.Reset() s.player.Play(reply) return nil } + +// dropBacklog resets the VAD and arranges for the frames captured during the +// round-trip to be thrown away as they arrive. +// +// Discarding them is also what keeps barge-in honest. The Frames guard is +// documented as "long enough that a door or a cough does not cut her off", +// which assumes the frames are real time. Draining a backlog delivers five +// frames in microseconds, so without this she could be cut off by audio +// recorded before she started speaking. +func (s *session) dropBacklog(start time.Time) { + s.vad.Reset() + s.loudFrames = 0 + s.recent = s.recent[:0] + if elapsed := s.now().Sub(start); elapsed > 0 { + s.discard = int(elapsed / frameDuration) + } +} diff --git a/cmd/mavwaked/session_test.go b/cmd/mavwaked/session_test.go index 8fa94d6..da6940b 100644 --- a/cmd/mavwaked/session_test.go +++ b/cmd/mavwaked/session_test.go @@ -5,6 +5,7 @@ import ( "errors" "math" "testing" + "time" "github.com/kami/maven/internal/audio" ) @@ -280,3 +281,147 @@ func TestBargeInConfigEnabled(t *testing.T) { } } } + +// slowSender models the real thing: a round-trip takes wall-clock time, and +// the microphone keeps recording into a pipe nobody is reading. +type slowSender struct { + fakeSender + clock *time.Time + took time.Duration +} + +func (s *slowSender) Send(ctx context.Context, utt audio.Audio, lang string) (audio.Audio, error) { + *s.clock = s.clock.Add(s.took) + return s.fakeSender.Send(ctx, utt, lang) +} + +// newSlowSession wires a session whose round-trip takes took of wall time. +func newSlowSession(barge bargeInConfig, reply audio.Audio, err error, took time.Duration) (*session, *fakePlayer, *slowSender) { + now := time.Unix(0, 0) + p := &fakePlayer{} + snd := &slowSender{fakeSender: fakeSender{reply: reply, err: err}, clock: &now, took: took} + sess := newSession(NewVAD(0, 0, 0, 0), p, snd, "ru", barge) + sess.now = func() time.Time { return now } + return sess, p, snd +} + +// A text-only turn starts no playback, so the half-duplex gate does not cover +// it. The backlog captured during the round-trip has to be dropped anyway, or +// three seconds of room arrives at pipe speed and becomes a command. +func TestSessionDropsBacklogAfterAnEmptyReply(t *testing.T) { + sess, p, snd := newSlowSession(bargeInConfig{}, audio.Audio{Format: audio.PCM16kMono}, nil, 3*time.Second) + + speakThenPause(t, sess) + if p.plays != 0 || len(snd.sent) != 1 { + t.Fatalf("plays = %d, sent = %d; want one text-only turn", p.plays, len(snd.sent)) + } + // The tail of speakThenPause already spent a couple of them. + if want := int(3 * time.Second / frameDuration); sess.discard+sess.dropped != want { + t.Fatalf("discard %d + dropped %d frames, want %d (3s of backlog)", sess.discard, sess.dropped, want) + } + + // The burst: the whole backlog, all of it him still talking. + loud := frameAt(0.35) + for i := 0; i < sess.discard; i++ { + if err := sess.feed(context.Background(), loud); err != nil { + t.Fatal(err) + } + } + if len(snd.sent) != 1 { + t.Errorf("the backlog was sent as a second utterance (sent = %d)", len(snd.sent)) + } + if sess.dropped == 0 { + t.Error("no frames were counted as backlog") + } +} + +// Same on the error path. A dead daemon used to seed the next spurious trigger +// on every failed turn, so a dead socket drove a retry loop off backlog alone. +func TestSessionDropsBacklogAfterASendError(t *testing.T) { + sess, _, _ := newSlowSession(bargeInConfig{}, audio.Audio{}, errors.New("boom"), 3*time.Second) + + // Not speakThenPause: the dispatch returns the send error, which that + // helper treats as fatal. + loud := frameAt(0.35) + for i := 0; i < (defaultSpeechMs+defaultFrameMs-1)/defaultFrameMs+5; i++ { + _ = sess.feed(context.Background(), loud) + } + for i := 0; i < (defaultSilenceMs+defaultFrameMs-1)/defaultFrameMs+2; i++ { + _ = sess.feed(context.Background(), silentBytes()) + } + if sess.discard == 0 { + t.Fatal("a failed round-trip left the backlog to be fed into the VAD") + } +} + +// Barge-in must not be triggerable by the backlog. Those frames are him +// finishing the sentence he started before she answered, delivered in +// microseconds, and the five-frame guard assumes real time. +func TestSessionBacklogCannotBargeIn(t *testing.T) { + sess, p, _ := newSlowSession(bargeInConfig{RMS: 0.12, Frames: 5}, replyAudio(), nil, 3*time.Second) + + speakThenPause(t, sess) + if p.plays != 1 || !p.Playing() { + t.Fatalf("plays = %d, playing = %v; want the reply playing", p.plays, p.Playing()) + } + + loud := frameAt(0.35) + backlog := sess.discard + if backlog < 5 { + t.Fatalf("discard = %d, want a real backlog", backlog) + } + for i := 0; i < backlog; i++ { + if err := sess.feed(context.Background(), loud); err != nil { + t.Fatal(err) + } + } + if p.stops != 0 { + t.Fatalf("she was cut off by audio recorded before she started speaking (stops = %d)", p.stops) + } + + // Real-time speech after the backlog still interrupts her. + for i := 0; i < 5; i++ { + if err := sess.feed(context.Background(), loud); err != nil { + t.Fatal(err) + } + } + if p.stops != 1 { + t.Fatalf("stops = %d, want 1 — barge-in must still work after the backlog", p.stops) + } +} + +// The frames that proved he was interrupting are replayed into the VAD, so his +// first word is not clipped. Five trigger frames plus five real ones reach the +// 300ms speech threshold; without the replay the first five are lost and no +// utterance is produced at all. +func TestSessionReplaysTheBargeInTriggerFrames(t *testing.T) { + sess, p, snd := newTestSession(bargeInConfig{RMS: 0.12, Frames: 5}) + speakThenPause(t, sess) + + veryLoud := frameAt(0.35) + for i := 0; i < 5; i++ { + _ = sess.feed(context.Background(), veryLoud) + } + if p.stops != 1 { + t.Fatalf("expected barge-in, stops = %d", p.stops) + } + if p.Playing() { + t.Fatal("fake player still playing after Stop") + } + + speechFrames := (defaultSpeechMs + defaultFrameMs - 1) / defaultFrameMs + for i := 0; i < speechFrames-5; i++ { + if err := sess.feed(context.Background(), veryLoud); err != nil { + t.Fatal(err) + } + } + silenceFrames := (defaultSilenceMs+defaultFrameMs-1)/defaultFrameMs + 2 + for i := 0; i < silenceFrames; i++ { + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatal(err) + } + } + if len(snd.sent) != 2 { + t.Fatalf("sent %d utterances, want 2 — the 150ms that triggered barge-in was clipped", len(snd.sent)) + } +}