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)
 

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

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

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