Merge pull request 'mavwaked registers as a voice consumer it cannot honor, so a spoken turn silences every nudge' (#218) from task/671-mavwaked-registers-as-a-voice-consumer-i into master

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

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

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

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) {

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

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
 	}

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