Maven/cmd/mavwaked/session_test.go

package main

import (
	"context"
	"errors"
	"math"
	"testing"
	"time"

	"github.com/kami/maven/internal/audio"
)

// fakePlayer records Play/Stop instead of shelling out to aplay.
type fakePlayer struct {
	playing bool
	plays   int
	stops   int
	last    audio.Audio
}

func (p *fakePlayer) Play(a audio.Audio) { p.playing = true; p.plays++; p.last = a }
func (p *fakePlayer) Stop()              { p.playing = false; p.stops++ }
func (p *fakePlayer) Playing() bool      { return p.playing }

// fakeSender records what was shipped and hands back a canned reply.
type fakeSender struct {
	sent  []audio.Audio
	reply audio.Audio
	err   error
}

func (s *fakeSender) Send(_ context.Context, utt audio.Audio, _ string) (audio.Audio, error) {
	s.sent = append(s.sent, utt)
	return s.reply, s.err
}

func replyAudio() audio.Audio {
	return audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 16000)}
}

// frameAt returns a 30ms frame whose RMS is approximately rms.
func frameAt(rms float64) []byte {
	amp := rms * math.Sqrt2 * 32768
	f := make([]int16, frameSamples)
	for i := range f {
		f[i] = int16(amp * math.Sin(2*math.Pi*440*float64(i)/16000))
	}
	return pcmBytes(f)
}

func silentBytes() []byte { return make([]byte, frameSamples*2) }

// newTestSession wires a session with fakes and a default VAD.
func newTestSession(barge bargeInConfig) (*session, *fakePlayer, *fakeSender) {
	p := &fakePlayer{}
	s := &fakeSender{reply: replyAudio()}
	return newSession(NewVAD(0, 0, 0, 0), p, s, "ru", barge), p, s
}

// speakThenPause drives a full utterance through the session: enough loud
// frames to trigger, then enough silence to end it.
func speakThenPause(t *testing.T, sess *session) {
	t.Helper()
	speechFrames := (defaultSpeechMs + defaultFrameMs - 1) / defaultFrameMs
	silenceFrames := (defaultSilenceMs+defaultFrameMs-1)/defaultFrameMs + 2
	loud := frameAt(0.35)
	for i := 0; i < speechFrames+5; i++ {
		if err := sess.feed(context.Background(), loud); err != nil {
			t.Fatalf("feed loud frame %d: %v", i, err)
		}
	}
	for i := 0; i < silenceFrames; i++ {
		if err := sess.feed(context.Background(), silentBytes()); err != nil {
			t.Fatalf("feed silent frame %d: %v", i, err)
		}
	}
}

func TestSessionSendsUtteranceAndPlaysReply(t *testing.T) {
	sess, p, snd := newTestSession(bargeInConfig{})
	speakThenPause(t, sess)

	if len(snd.sent) != 1 {
		t.Fatalf("sent %d utterances, want 1", len(snd.sent))
	}
	if snd.sent[0].Format != audio.PCM16kMono {
		t.Errorf("utterance format = %+v, want canonical", snd.sent[0].Format)
	}
	if p.plays != 1 {
		t.Errorf("plays = %d, want 1", p.plays)
	}
}

// The bug this whole file exists for: while the speaker is running, the mic
// hears Maven and the old code shipped that back as a fresh command.
func TestSessionDoesNotHearItselfWhilePlaying(t *testing.T) {
	sess, p, snd := newTestSession(bargeInConfig{})
	speakThenPause(t, sess)
	if !p.Playing() {
		t.Fatal("expected playback to be running after the reply")
	}

	// Feed a long stretch of loud audio — Maven's own voice coming back in.
	base := sess.suppressed
	loud := frameAt(0.35)
	for i := 0; i < 200; i++ {
		if err := sess.feed(context.Background(), loud); err != nil {
			t.Fatalf("feed echo frame %d: %v", i, err)
		}
	}

	if len(snd.sent) != 1 {
		t.Fatalf("sent %d utterances, want 1 — her own reply was captured as a command", len(snd.sent))
	}
	if got := sess.suppressed - base; got != 200 {
		t.Errorf("suppressed %d of the 200 echo frames, want all of them", got)
	}
	if p.stops != 0 {
		t.Errorf("stops = %d, want 0 — barge-in is off, nothing should cut her off", p.stops)
	}
}

// With barge-in off, no amount of noise stops playback.
func TestSessionBargeInDisabledByDefault(t *testing.T) {
	sess, p, _ := newTestSession(bargeInConfig{})
	if sess.barge.Enabled() {
		t.Fatal("zero bargeInConfig must be disabled")
	}
	speakThenPause(t, sess)
	veryLoud := frameAt(0.6)
	for i := 0; i < 50; i++ {
		_ = sess.feed(context.Background(), veryLoud)
	}
	if p.stops != 0 || sess.bargeIns != 0 {
		t.Fatalf("stops = %d, bargeIns = %d, want 0 with barge-in off", p.stops, sess.bargeIns)
	}
}

func TestSessionBargeInCutsPlayback(t *testing.T) {
	barge := bargeInConfig{RMS: 0.12, Frames: 5}
	sess, p, _ := newTestSession(barge)
	speakThenPause(t, sess)
	if !p.Playing() {
		t.Fatal("expected playback after the reply")
	}

	// Four loud frames must not be enough — a door closing is not a voice.
	veryLoud := frameAt(0.35)
	for i := 0; i < 4; i++ {
		_ = sess.feed(context.Background(), veryLoud)
	}
	if p.stops != 0 {
		t.Fatalf("playback cut after 4 frames, want it to hold until %d", barge.Frames)
	}

	// The fifth cuts her off.
	_ = sess.feed(context.Background(), veryLoud)
	if p.stops != 1 || sess.bargeIns != 1 {
		t.Fatalf("stops = %d, bargeIns = %d, want 1 and 1", p.stops, sess.bargeIns)
	}
	if p.Playing() {
		t.Fatal("still playing after barge-in")
	}
}

// A burst that falls back under the threshold resets the counter, so noise
// spread over a whole reply never accumulates into a false barge-in.
func TestSessionBargeInNeedsConsecutiveFrames(t *testing.T) {
	sess, p, _ := newTestSession(bargeInConfig{RMS: 0.12, Frames: 5})
	speakThenPause(t, sess)

	veryLoud := frameAt(0.35)
	quiet := frameAt(0.02)
	for i := 0; i < 20; i++ {
		_ = sess.feed(context.Background(), veryLoud)
		_ = sess.feed(context.Background(), veryLoud)
		_ = sess.feed(context.Background(), quiet)
	}
	if p.stops != 0 || sess.bargeIns != 0 {
		t.Fatalf("stops = %d, bargeIns = %d, want 0 — two-frame bursts must not accumulate", p.stops, sess.bargeIns)
	}
}

// Speaker leak sits near the room floor; it must never reach the barge-in bar.
func TestSessionEchoLevelAudioNeverBargesIn(t *testing.T) {
	sess, p, _ := newTestSession(bargeInConfig{RMS: 0.12, Frames: 5})
	speakThenPause(t, sess)

	base := sess.suppressed
	leak := frameAt(0.05) // loud enough for the VAD, far under the barge bar
	for i := 0; i < 300; i++ {
		_ = sess.feed(context.Background(), leak)
	}
	if p.stops != 0 {
		t.Fatalf("stops = %d, want 0 — speaker leak must not read as barge-in", p.stops)
	}
	if got := sess.suppressed - base; got != 300 {
		t.Errorf("suppressed %d of the 300 leak frames, want all of them", got)
	}
}

// After barge-in the VAD must start clean, so the interrupting speech is
// captured as a whole utterance rather than joined onto echo state.
func TestSessionCapturesTheInterruptingUtterance(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)
	}

	// He keeps talking; that is a new command.
	speakThenPause(t, sess)
	if len(snd.sent) != 2 {
		t.Fatalf("sent %d utterances, want 2 — the interruption itself must be heard", len(snd.sent))
	}
	if p.plays != 2 {
		t.Errorf("plays = %d, want 2", p.plays)
	}
}

// A failed round-trip must surface as an error and must not start playback.
func TestSessionSendErrorDoesNotPlay(t *testing.T) {
	p := &fakePlayer{}
	snd := &fakeSender{err: errors.New("boom")}
	sess := newSession(NewVAD(0, 0, 0, 0), p, snd, "ru", bargeInConfig{})

	speechFrames := (defaultSpeechMs + defaultFrameMs - 1) / defaultFrameMs
	silenceFrames := (defaultSilenceMs+defaultFrameMs-1)/defaultFrameMs + 2
	loud := frameAt(0.35)
	var lastErr error
	for i := 0; i < speechFrames+5; i++ {
		_ = sess.feed(context.Background(), loud)
	}
	for i := 0; i < silenceFrames; i++ {
		if err := sess.feed(context.Background(), silentBytes()); err != nil {
			lastErr = err
		}
	}
	if lastErr == nil {
		t.Fatal("send error was swallowed")
	}
	if p.plays != 0 || p.Playing() {
		t.Fatalf("plays = %d, playing = %v, want no playback on a failed round-trip", p.plays, p.Playing())
	}
}

// An empty reply (text-only turn) must leave the capture side open.
func TestSessionEmptyReplyLeavesCaptureOpen(t *testing.T) {
	p := &fakePlayer{}
	snd := &fakeSender{reply: audio.Audio{Format: audio.PCM16kMono}}
	sess := newSession(NewVAD(0, 0, 0, 0), p, snd, "ru", bargeInConfig{})

	speakThenPause(t, sess)
	if p.plays != 0 {
		t.Fatalf("plays = %d, want 0 for an empty reply", p.plays)
	}
	speakThenPause(t, sess)
	if len(snd.sent) != 2 {
		t.Fatalf("sent %d, want 2 — capture must stay open when there is no audio reply", len(snd.sent))
	}
}

func TestBargeInConfigEnabled(t *testing.T) {
	cases := []struct {
		c    bargeInConfig
		want bool
	}{
		{bargeInConfig{}, false},
		{bargeInConfig{RMS: 0.12}, false},
		{bargeInConfig{Frames: 5}, false},
		{bargeInConfig{RMS: 0.12, Frames: 5}, true},
	}
	for _, tc := range cases {
		if got := tc.c.Enabled(); got != tc.want {
			t.Errorf("%+v.Enabled() = %v, want %v", tc.c, got, tc.want)
		}
	}
}

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