diff --git a/cmd/mavwaked/main.go b/cmd/mavwaked/main.go index f1bfa26..4f0c5a4 100644 --- a/cmd/mavwaked/main.go +++ b/cmd/mavwaked/main.go @@ -17,6 +17,11 @@ // commands at L0 (no destructive acts), which is what makes an accidental // trigger safe rather than expensive. // +// The conn carries both directions. mavwaked sends utterances and receives +// proactive nudges on it, and it is opened at startup rather than at the first +// utterance, because mavend registers a voice session on accept. See nudge.go +// for why a nudge that is not heard is worse than one that is not delivered. +// // 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 above @@ -86,7 +91,8 @@ func run(args []string) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) defer stop() - // Voice client — reused across utterances; SendRequest reconnects on error. + // Voice client — one conn carrying both directions. SendRequest reconnects + // on error, and the push receiver redials on its own clock. vc := voice.Dial(*addr) defer vc.Close() @@ -165,6 +171,14 @@ func run(args []string) error { } sess := newSession(vad, newAplayPlayer(), &voiceSender{vc: vc}, *lang, barge) + // Listen for nudges alongside capture. Connect eagerly so mavend has a + // voice session before he has said anything: without one, a nudge routed + // to voice finds nobody home and goes to the away channels instead. + if err := vc.Connect(ctx); err != nil { + log.Printf("mavwaked: voice server not reachable yet, retrying in background: %v", err) + } + go runNudgeReceiver(ctx, vc, sess) + return captureLoop(ctx, src, sess) } diff --git a/cmd/mavwaked/nudge.go b/cmd/mavwaked/nudge.go new file mode 100644 index 0000000..44f8ef8 --- /dev/null +++ b/cmd/mavwaked/nudge.go @@ -0,0 +1,79 @@ +package main + +// The receiving half of the voice reach (V-671). +// +// mavwaked used to send and never listen. It wired no PushHandler, and +// SendRequest discards a push frame when there is none. The consequence was +// not a missing feature but a silent one: mavend routes a nudge to the voice +// session that spoke most recently, and once mavwaked had spoken once it WAS +// that session. PushToMostRecent succeeded, the dispatcher counted the nudge +// delivered and stopped rerouting to telegram and ntfy, and mavwaked threw the +// audio away. He heard nothing, anywhere. +// +// So the connection is opened at startup rather than at the first utterance, +// and it is held open. A client that has never connected has no session, and +// the dispatcher must be able to tell "he is not at the machine" from "he is, +// and she has nothing to say". + +import ( + "context" + "encoding/json" + "log" + "time" + + "github.com/kami/maven/internal/voice" +) + +// nudgeRetry is how long to wait before dialling again after the conn ends. +// mavend restarts on every deploy, and a listener that gives up then is a +// listener that is deaf until the next reboot. +const nudgeRetry = 5 * time.Second + +// nudgeHandler decodes a push and hands the audio to the session, which +// speaks it through the same player the reply path uses. It does not play +// anything itself: the half-duplex gate and barge-in live on the capture +// loop, and a nudge has to sit under both. +type nudgeHandler struct{ sess *session } + +func (h *nudgeHandler) OnPush(p voice.Push) { + if p.Kind != voice.PushKindAudioNudge { + log.Printf("mavwaked: ignoring push of unknown kind %q", p.Kind) + return + } + var ap voice.AudioNudgePush + if err := json.Unmarshal(p.Params, &ap); err != nil { + log.Printf("mavwaked: nudge: decode: %v", err) + return + } + log.Printf("mavwaked: nudge from rule %q (severity %d): %q (%.2fs audio)", + ap.RuleName, ap.Severity, ap.Text, ap.Audio.Duration()) + if len(ap.Audio.Bytes) == 0 { + // mavttsd was down or the text was empty. Say so rather than going + // quiet: the dispatcher already counted this one as delivered. + log.Printf("mavwaked: nudge %q carried no audio, nothing to speak", ap.RuleName) + return + } + h.sess.Nudge(ap.Audio) +} + +// runNudgeReceiver keeps a push handler wired for as long as ctx lives, +// redialling whenever the conn ends. Returns when ctx is cancelled. +func runNudgeReceiver(ctx context.Context, vc *voice.Client, sess *session) { + h := &nudgeHandler{sess: sess} + for { + err := vc.RunPushReceiver(ctx, h) + if ctx.Err() != nil { + return + } + if err != nil { + log.Printf("mavwaked: nudge receiver: %v", err) + } else { + log.Printf("mavwaked: voice connection ended, reconnecting in %s", nudgeRetry) + } + select { + case <-ctx.Done(): + return + case <-time.After(nudgeRetry): + } + } +} diff --git a/cmd/mavwaked/nudge_test.go b/cmd/mavwaked/nudge_test.go new file mode 100644 index 0000000..20102af --- /dev/null +++ b/cmd/mavwaked/nudge_test.go @@ -0,0 +1,170 @@ +package main + +// The receiving half: a nudge pushed by mavend has to reach the speaker, and +// it has to obey the same two gates a reply obeys (V-671). + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/voice" +) + +func nudgeAudio() audio.Audio { + return audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 8000)} +} + +// pushFrame builds the frame mavend's voicesink sends. +func pushFrame(t *testing.T, a audio.Audio) voice.Push { + t.Helper() + body, err := json.Marshal(voice.AudioNudgePush{ + RuleName: "test-rule", + Severity: 3, + Audio: a, + Text: "пора пить воду", + Ts: time.Unix(0, 0), + }) + if err != nil { + t.Fatalf("marshal push: %v", err) + } + return voice.Push{Kind: voice.PushKindAudioNudge, Params: body} +} + +// The defect itself: the push arrived and nothing came out of the speaker. +func TestNudgeReachesThePlayer(t *testing.T) { + sess, p, snd := newTestSession(bargeInConfig{}) + (&nudgeHandler{sess: sess}).OnPush(pushFrame(t, nudgeAudio())) + + if p.plays != 0 { + t.Fatal("nudge played from the push goroutine; it must wait for the capture loop") + } + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != 1 { + t.Fatalf("plays = %d, want 1", p.plays) + } + if len(p.last.Bytes) != 8000 { + t.Errorf("played %d bytes, want the nudge audio", len(p.last.Bytes)) + } + if sess.nudges != 1 { + t.Errorf("nudges = %d, want 1", sess.nudges) + } + if len(snd.sent) != 0 { + t.Errorf("a nudge must not be shipped back to the daemon as an utterance") + } +} + +// A push of some other kind, or one carrying no audio, must not reach the +// player and must not wedge the one that follows. +func TestNudgeIgnoresUnusablePushes(t *testing.T) { + sess, p, _ := newTestSession(bargeInConfig{}) + h := &nudgeHandler{sess: sess} + + h.OnPush(voice.Push{Kind: "something-else", Params: json.RawMessage(`{}`)}) + h.OnPush(voice.Push{Kind: voice.PushKindAudioNudge, Params: json.RawMessage(`not json`)}) + h.OnPush(pushFrame(t, audio.Audio{Format: audio.PCM16kMono})) + + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != 0 { + t.Fatalf("plays = %d, want 0", p.plays) + } + + h.OnPush(pushFrame(t, nudgeAudio())) + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != 1 { + t.Fatalf("plays after a usable nudge = %d, want 1", p.plays) + } +} + +// The half-duplex gate covers a nudge exactly as it covers a reply: she does +// not start one over herself, and the mic stays muted while it runs. +func TestNudgeWaitsForTheReplyToFinish(t *testing.T) { + sess, p, _ := newTestSession(bargeInConfig{}) + speakThenPause(t, sess) + if !p.Playing() { + t.Fatal("expected the reply to be playing") + } + plays := p.plays + + (&nudgeHandler{sess: sess}).OnPush(pushFrame(t, nudgeAudio())) + for i := 0; i < 20; i++ { + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + } + if p.plays != plays { + t.Fatalf("nudge cut across the reply: plays = %d, want %d", p.plays, plays) + } + + p.playing = false + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != plays+1 { + t.Fatalf("nudge never played after the reply ended: plays = %d", p.plays) + } +} + +// Speaking a nudge must not leave half a sentence in the VAD. The frames +// captured before it are pre-nudge speech, and splicing them onto whatever he +// says afterwards ships one utterance that is two. +func TestNudgeResetsTheVAD(t *testing.T) { + sess, p, snd := newTestSession(bargeInConfig{}) + loud := frameAt(0.35) + speechFrames := (defaultSpeechMs + defaultFrameMs - 1) / defaultFrameMs + for i := 0; i < speechFrames+5; i++ { + if err := sess.feed(context.Background(), loud); err != nil { + t.Fatalf("feed: %v", err) + } + } + + (&nudgeHandler{sess: sess}).OnPush(pushFrame(t, nudgeAudio())) + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != 1 { + t.Fatalf("nudge did not play: plays = %d", p.plays) + } + + // Playback ends, silence follows. The half-formed utterance must be gone + // rather than closing on the first quiet frame. + p.playing = false + silenceFrames := (defaultSilenceMs+defaultFrameMs-1)/defaultFrameMs + 2 + for i := 0; i < silenceFrames; i++ { + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + } + if len(snd.sent) != 0 { + t.Fatalf("sent %d utterances after a nudge, want 0", len(snd.sent)) + } +} + +// Two nudges queued back to back: the newer one is what he hears. The +// PushHandler contract in internal/voice says the next nudge replaces the +// stale one rather than dogpiling on it. +func TestNudgeReplacesAnUnspokenOne(t *testing.T) { + sess, p, _ := newTestSession(bargeInConfig{}) + h := &nudgeHandler{sess: sess} + + h.OnPush(pushFrame(t, audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 4000)})) + h.OnPush(pushFrame(t, audio.Audio{Format: audio.PCM16kMono, Bytes: make([]byte, 12000)})) + + if err := sess.feed(context.Background(), silentBytes()); err != nil { + t.Fatalf("feed: %v", err) + } + if p.plays != 1 { + t.Fatalf("plays = %d, want 1", p.plays) + } + if len(p.last.Bytes) != 12000 { + t.Errorf("played %d bytes, want the newer nudge", len(p.last.Bytes)) + } +} diff --git a/cmd/mavwaked/session.go b/cmd/mavwaked/session.go index 24de887..8c0d600 100644 --- a/cmd/mavwaked/session.go +++ b/cmd/mavwaked/session.go @@ -7,6 +7,7 @@ package main import ( "context" "log" + "sync" "time" "github.com/kami/maven/internal/audio" @@ -62,11 +63,19 @@ type session struct { // whenever playback ends. loudFrames int + // pending holds a nudge the push receiver handed over, waiting for the + // capture loop to speak it. It is the one field written from another + // goroutine, hence the mutex; everything else in this struct belongs to + // the capture loop alone. + nudgeMu sync.Mutex + pending *audio.Audio + // 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 + nudges int // proactive pushes spoken through the speaker // loudSum and loudSeen accumulate the energy of suppressed frames, so // the operator can read what the room actually measures and set @@ -150,6 +159,10 @@ func (s *session) feed(ctx context.Context, frame []byte) error { s.vad.Reset() } + if s.startPendingNudge() { + return nil + } + utt, state := s.vad.Feed(PCMToI16(frame)) if state == StateSpeech || utt.Bytes == nil { return nil @@ -157,6 +170,54 @@ func (s *session) feed(ctx context.Context, frame []byte) error { return s.dispatch(ctx, utt) } +// Nudge hands proactive audio to the session, to be spoken as soon as the +// capture loop finds a quiet moment. Safe to call from the push receiver +// goroutine; nothing else here is. +// +// A nudge arriving while one is already waiting REPLACES it. That is the +// contract internal/voice states for PushHandler: the next nudge replaces the +// stale one in his attention rather than dogpiling on it. +func (s *session) Nudge(a audio.Audio) { + if len(a.Bytes) == 0 { + return + } + s.nudgeMu.Lock() + if s.pending != nil { + log.Printf("mavwaked: nudge replaced one still waiting to be spoken") + } + s.pending = &a + s.nudgeMu.Unlock() +} + +// takeNudge removes and returns the waiting nudge, or nil. +func (s *session) takeNudge() *audio.Audio { + s.nudgeMu.Lock() + defer s.nudgeMu.Unlock() + a := s.pending + s.pending = nil + return a +} + +// startPendingNudge speaks a waiting nudge and reports whether it started +// one. It runs on the capture loop, past the half-duplex gate, so a nudge +// never cuts across a reply and never plays into a backlog drain. +// +// The VAD is reset first. Playback is about to suppress every frame until it +// ends, and a half-heard sentence left in the VAD would splice onto whatever +// he says afterwards. Barge-in needs no special case: it reads the player, +// and the player does not care which audio it is playing. +func (s *session) startPendingNudge() bool { + a := s.takeNudge() + if a == nil { + return false + } + s.vad.Reset() + s.nudges++ + log.Printf("mavwaked: speaking nudge (%.2fs audio)", a.Duration()) + s.player.Play(*a) + return true +} + // keepRecent stores a copy of one barge-in trigger frame, keeping at most // barge.Frames of them. func (s *session) keepRecent(frame []byte) { diff --git a/docs/deployment.md b/docs/deployment.md index 19482f3..5bf76d1 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -170,9 +170,13 @@ The device is `plughw:0,0` and not `hw:0,0`. The fifine offers 2 channels at 44100 or 48000 and nothing else, and mavwaked asks arecord for 16kHz mono. Bare `hw` dies on "Channels count non available" before a frame is read. -There is no wake word yet (V-487 stage two), so the loop runs open. mavwaked -connects lazily, so `voicesink` cannot push a nudge to it until it has sent one -utterance. +There is no wake word yet (V-487 stage two), so the loop runs open. + +mavwaked connects at startup and holds the conn, so a nudge routed to voice +reaches the speaker before he has said anything (V-671). It used to connect +lazily, which made the failure silent rather than absent: after one utterance +the session existed, `PushToMostRecent` succeeded, the dispatcher stopped +rerouting to telegram and ntfy, and mavwaked discarded the audio. **Passwords are read from files, never taken as flag values.** `mavcaldav` uses `-pass-file` and `-render-pass-file`. `mavpoll` and `mavmaild` follow the same diff --git a/internal/voice/client.go b/internal/voice/client.go index ca7260b..c8bf8c0 100644 --- a/internal/voice/client.go +++ b/internal/voice/client.go @@ -9,15 +9,18 @@ // // The wire is symmetric: a Request from the client is answered by a // Response with a matching ID, OR a server-initiated Push frame (no ID) -// may arrive interleaved. SendRequest loops reading frames, drops Push -// frames to the harness if a receiver is running (or silently if not), -// and returns the first Response with the matching ID. +// may arrive interleaved. One reader goroutine per connection owns the +// socket. It hands each Response to whichever SendRequest is waiting on +// that ID and each Push to the handler, so a client may send and listen +// at the same time on one conn. mavwaked needs exactly that: it speaks +// utterances and it must hear nudges, and the server routes a nudge to +// the session that spoke most recently, so a second listening conn would +// never be picked (V-671). package voice import ( "context" "encoding/json" - "errors" "fmt" "io" "net" @@ -41,13 +44,22 @@ type PushHandler interface { // Client — one connection to the voice.Server. type Client struct { addr string - mu sync.Mutex - c net.Conn nextID atomic.Uint64 - // pushCh fan-out: a reader goroutine (started by RunPushReceiver) - // writes Push frames here; SendRequest also drains it when no reader - // is running (drops the frame in that case). + mu sync.Mutex + c net.Conn + // pending holds one channel per in-flight request, keyed by frame id. + // The reader goroutine delivers the Response here and deletes the entry. + pending map[uint64]chan *Response + // dead is closed by the reader goroutine when this conn ends, so a + // waiting SendRequest fails at once instead of at its own deadline. + dead chan struct{} + + // wmu serialises writes. Frames must not interleave on the wire. + wmu sync.Mutex + + // pushH is set by RunPushReceiver and survives a reconnect, because the + // client that wants pushes wants them on whatever conn it ends up with. pushMu sync.Mutex pushH PushHandler } @@ -55,6 +67,15 @@ type Client struct { // Dial returns a Client that will connect to addr on first use. func Dial(addr string) *Client { return &Client{addr: addr} } +// Connect opens the conn now rather than on the first request. mavwaked calls +// it at startup: the server registers a session on accept, and a client that +// has never connected cannot be sent a nudge. +func (c *Client) Connect(ctx context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + return c.ensureConnLocked(ctx) +} + // Close releases the conn. Idempotent. func (c *Client) Close() error { c.mu.Lock() @@ -75,12 +96,14 @@ func (c *Client) PushToTalk(ctx context.Context, a audio.Audio, lang string) (Pu return out, err } +// requestTimeout bounds a round-trip with no deadline on its context. It is +// generous because the far end runs speech-to-text, a router and a voice. +const requestTimeout = 120 * time.Second + // SendRequest sends one Request frame and waits for the matching Response. -// Push frames received while waiting are dropped on the floor UNLESS a -// PushHandler has been wired via RunPushReceiver, in which case the handler -// is invoked inline (still synchronous with the SendRequest caller's -// read). For sanity, the reference client runs either one-shot (no -// receiver) or interactive (RunPushReceiver, no concurrent SendRequest). +// Push frames arriving meanwhile go to the handler on the reader goroutine, +// so listening and sending on one Client is supported rather than merely +// tolerated. func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any) error { body, err := marshalParams(params) if err != nil { @@ -94,58 +117,69 @@ func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any) c.mu.Unlock() return err } - conn := c.c + conn, dead := c.c, c.dead + ch := make(chan *Response, 1) + c.pending[id] = ch c.mu.Unlock() - if dl, ok := ctx.Deadline(); ok { - _ = conn.SetDeadline(dl) - } else { - _ = conn.SetDeadline(time.Now().Add(120 * time.Second)) - } - defer conn.SetDeadline(time.Time{}) - - if err := writeFrame(conn, &req); err != nil { - c.teardown() + c.wmu.Lock() + err = writeFrame(conn, &req) + c.wmu.Unlock() + if err != nil { + c.forget(id) + c.teardownConn(conn) return err } - for { - resp, push, err := readOneFrame(conn) - if err != nil { - c.teardown() - return err - } - if push != nil { - c.deliverPush(*push) - continue - } - if resp.ID != id { - continue // not ours; ignore (singleplex ⇒ shouldn't happen) - } - if resp.Error != nil { - return hydrate(resp.Error) - } - if out != nil { - if err := json.Unmarshal(resp.Result, out); err != nil { - return fmt.Errorf("voice: unmarshal result: %w", err) - } - } - return nil + + timer := time.NewTimer(requestTimeout) + defer timer.Stop() + + var resp *Response + select { + case resp = <-ch: + case <-dead: + c.forget(id) + return fmt.Errorf("voice: connection closed before reply") + case <-ctx.Done(): + c.forget(id) + return ctx.Err() + case <-timer.C: + c.forget(id) + c.teardownConn(conn) + return fmt.Errorf("voice: no reply within %s", requestTimeout) } + + if resp.Error != nil { + return hydrate(resp.Error) + } + if out != nil { + if err := json.Unmarshal(resp.Result, out); err != nil { + return fmt.Errorf("voice: unmarshal result: %w", err) + } + } + return nil } -// RunPushReceiver spawns a reader goroutine that delivers Push frames to h -// until the conn closes or Close is called. Today's reference client uses -// this in -listen mode (proactive voice playback). SendRequest and -// RunPushReceiver SHOULD NOT be used concurrently on the same Client — the -// wire is singleplex at the reference client's scale; production picks one -// mode per conn. Returns when the goroutine ends (ctx cancel or conn close). +// forget drops an abandoned request so a late Response is discarded rather +// than delivered to nobody. +func (c *Client) forget(id uint64) { + c.mu.Lock() + delete(c.pending, id) + c.mu.Unlock() +} + +// RunPushReceiver wires h and blocks until the conn ends or ctx is +// cancelled. Frames are read by the per-conn reader goroutine, so a client +// may call SendRequest on the same Client while this is running. Returns nil +// when the conn ended, so a caller that wants to stay reachable reconnects +// and calls it again. func (c *Client) RunPushReceiver(ctx context.Context, h PushHandler) error { c.mu.Lock() if err := c.ensureConnLocked(ctx); err != nil { c.mu.Unlock() return err } - conn := c.c + dead := c.dead c.mu.Unlock() c.pushMu.Lock() @@ -158,21 +192,34 @@ func (c *Client) RunPushReceiver(ctx context.Context, h PushHandler) error { c.pushMu.Unlock() }() + select { + case <-ctx.Done(): + return ctx.Err() + case <-dead: + return nil + } +} + +// readLoop owns conn for its whole life. It ends on any read error, which is +// how a closed conn, a killed server and a cancelled dial all arrive here. +func (c *Client) readLoop(conn net.Conn, dead chan struct{}) { + defer close(dead) for { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - _, push, err := readOneFrame(conn) + resp, push, err := readOneFrame(conn) if err != nil { - if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { - return nil - } - return err + c.teardownConn(conn) + return } if push != nil { c.deliverPush(*push) + continue + } + c.mu.Lock() + ch := c.pending[resp.ID] + delete(c.pending, resp.ID) + c.mu.Unlock() + if ch != nil { + ch <- resp } } } @@ -196,13 +243,19 @@ func (c *Client) ensureConnLocked(ctx context.Context) error { return fmt.Errorf("voice: dial %s: %w", c.addr, err) } c.c = conn + c.pending = make(map[uint64]chan *Response) + c.dead = make(chan struct{}) + go c.readLoop(conn, c.dead) return nil } -func (c *Client) teardown() { +// teardownConn closes conn and forgets it, but only if it is still the live +// one. A reconnect may already have replaced it, and closing the new conn +// because the old one died takes the client down on every hiccup. +func (c *Client) teardownConn(conn net.Conn) { c.mu.Lock() defer c.mu.Unlock() - if c.c != nil { + if c.c != nil && c.c == conn { _ = c.c.Close() c.c = nil } diff --git a/internal/voice/voice_test.go b/internal/voice/voice_test.go index 3483352..57b77e5 100644 --- a/internal/voice/voice_test.go +++ b/internal/voice/voice_test.go @@ -221,3 +221,113 @@ func TestClientListenModeReceivesPush(t *testing.T) { type pushHandlerFunc func(Push) func (f pushHandlerFunc) OnPush(p Push) { f(p) } + +// The whole point of the per-conn reader (V-671): mavwaked speaks utterances +// and must hear nudges, and the server routes a nudge to the session that +// spoke most recently. A second listening conn would never be picked, so both +// directions have to share one conn. +func TestClientSendsAndListensOnOneConn(t *testing.T) { + l := newTestListener(t) + sess := NewSessions() + h := &stubHandler{} + srv := NewServer(l.Addr().String(), h, sess) + if err := srv.Listen(); err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { + _ = srv.Close() + waitPort() + }() + go func() { _ = srv.Serve() }() + + c := Dial(l.Addr().String()) + defer c.Close() + + got := make(chan audio.Audio, 4) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + _ = c.RunPushReceiver(ctx, pushHandlerFunc(func(p Push) { + var ap AudioNudgePush + if err := json.Unmarshal(p.Params, &ap); err == nil { + got <- ap.Audio + } + })) + }() + + for i := 0; i < 100 && sess.Active() < 1; i++ { + time.Sleep(10 * time.Millisecond) + } + if sess.Active() != 1 { + t.Fatalf("active sessions = %d, want exactly 1", sess.Active()) + } + + // A round-trip while the receiver is running. Before the reader owned the + // conn, this and the receiver raced for every frame. + resp, err := c.PushToTalk(context.Background(), audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("hello")}, "ru") + if err != nil { + t.Fatalf("PushToTalk with a receiver running: %v", err) + } + if resp.ReplyText != "got it" { + t.Fatalf("ReplyText = %q, want %q", resp.ReplyText, "got it") + } + + // And the nudge still arrives, on the session that just spoke. + err = sess.PushToMostRecent(context.Background(), AudioNudgePush{ + RuleName: "after-speaking", + Audio: audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("proactive")}, + }) + if err != nil { + t.Fatalf("PushToMostRecent: %v", err) + } + select { + case a := <-got: + if string(a.Bytes) != "proactive" { + t.Fatalf("received %q, want the nudge audio", string(a.Bytes)) + } + case <-time.After(2 * time.Second): + t.Fatal("nudge never reached the handler after the client had spoken") + } +} + +// A request abandoned by its context must not leave its slot behind, or a +// long-running client leaks one channel per timeout. +func TestClientForgetsAbandonedRequests(t *testing.T) { + l := newTestListener(t) + sess := NewSessions() + srv := NewServer(l.Addr().String(), &blockingHandler{}, sess) + if err := srv.Listen(); err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { + _ = srv.Close() + waitPort() + }() + go func() { _ = srv.Serve() }() + + c := Dial(l.Addr().String()) + defer c.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, err := c.PushToTalk(ctx, audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("x")}, "ru") + if err == nil { + t.Fatal("expected the round-trip to fail on its context") + } + + c.mu.Lock() + n := len(c.pending) + c.mu.Unlock() + if n != 0 { + t.Fatalf("pending = %d after an abandoned request, want 0", n) + } +} + +// blockingHandler never answers, so the client's context is what ends the +// round-trip. +type blockingHandler struct{} + +func (blockingHandler) HandlePushToTalk(ctx context.Context, _ PushToTalkReq, _ uint64) (PushToTalkResp, error) { + <-ctx.Done() + return PushToTalkResp{}, ctx.Err() +}