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