From 1f1e00278958a26aee1f8171909e35afbf887174 Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 9 Aug 2026 13:44:52 +0400 Subject: [PATCH 1/4] One reader goroutine per voice conn, so a client can send and listen (V-671) SendRequest and RunPushReceiver each read the conn, so a client that wanted both raced for every frame. A second listening conn is not the fix: it never sends a request, so its lastActive never moves and PushToMostRecent never picks it. mavwaked needs both on one conn. The reader now owns the socket for the life of the conn. It hands each Response to whichever SendRequest waits on that id, and each Push to the handler. SendRequest waits on its own channel, on the conn dying, on its context, or on a timeout, and forgets its slot on every path that leaves without an answer. RunPushReceiver just wires the handler and blocks. Connect opens the conn without sending anything, for a client that must hold a session before it has spoken. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013ptwopxyo3Z2kwFckHkLvN --- internal/voice/client.go | 183 ++++++++++++++++++++++------------- internal/voice/voice_test.go | 110 +++++++++++++++++++++ 2 files changed, 228 insertions(+), 65 deletions(-) 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() +} -- 2.52.0 From 9c7bafd5b1866c0b790bd064e9ced52ee481f2cc Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 9 Aug 2026 13:45:04 +0400 Subject: [PATCH 2/4] Let mavwaked hear the nudges it was already being sent (V-671) It wired no PushHandler, and SendRequest discards a push frame when there is none. That was not a missing feature but a silent one. mavend routes a nudge to the voice session that spoke most recently, so once mavwaked had spoken once it WAS that session. PushToMostRecent succeeded, the dispatcher counted the nudge delivered and stopped rerouting to the away channels, and mavwaked threw the audio away. He heard nothing, anywhere. It now connects at startup rather than at the first utterance, because the dispatcher has to tell "he is not at the machine" from "he is, and she has nothing to say". The receiver redials on its own clock, since mavend restarts on every deploy. A nudge is queued, not played where it arrives. The capture loop picks it up on the next frame, so the half-duplex gate and barge-in cover it the way they cover a reply. It resets the VAD first: playback is about to suppress every frame, and a half-heard sentence would otherwise splice onto whatever he says next. A nudge arriving while one still waits replaces it, which is the contract internal/voice states for PushHandler. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013ptwopxyo3Z2kwFckHkLvN --- cmd/mavwaked/main.go | 16 ++++++++- cmd/mavwaked/nudge.go | 79 +++++++++++++++++++++++++++++++++++++++++ cmd/mavwaked/session.go | 61 +++++++++++++++++++++++++++++++ 3 files changed, 155 insertions(+), 1 deletion(-) create mode 100644 cmd/mavwaked/nudge.go 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/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) { -- 2.52.0 From d0ea927ac322461e31e27eff359025f1813552f3 Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 9 Aug 2026 13:45:04 +0400 Subject: [PATCH 3/4] Pin the five things a nudge must do at the speaker (V-671) It reaches the player, but not from the push goroutine. An unusable push is dropped and does not wedge the next one. It waits for a reply to finish. It resets the VAD, so the frames before it are not spliced onto what he says after. And a second nudge replaces an unspoken first. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013ptwopxyo3Z2kwFckHkLvN --- cmd/mavwaked/nudge_test.go | 170 +++++++++++++++++++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 cmd/mavwaked/nudge_test.go 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)) + } +} -- 2.52.0 From 8c30971a96e97bde4fcc64415bde6b9769fb3a2a Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 9 Aug 2026 13:45:20 +0400 Subject: [PATCH 4/4] Say that mavwaked now holds the conn from startup (V-671) The lazy-connect note is no longer true and the trap it described was the opposite way round: the session existed and the audio was discarded. diff-budget.sh blocks the branch at 615 changed lines. This commit is markdown only, which the repo's own pre-commit hook exempts, and it corrects a line the code in this branch has just falsified. --- docs/deployment.md | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) 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 -- 2.52.0