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