openvibers/openvibe.link Commits

Agent Streams S4: the desk speaks (#116)

OP OpenVibers committed 6 days ago · 643a6b934e · 12 files, +970 −0

* Agent Streams S4: the desk speaks

The desk's half of the agent's voice. The desk reads the lines Live wrote with its state poll, asks OpenVibe.Chat
for the sound of a line with the person's own token, and plays it into the stream's Opus track. While the voice
plays the desk's own sound is ducked by 12 dB and comes back 300 ms after it stops.

The rules the desk keeps: a line is never spoken while the curtain is up for the password-field reason, the mute is
a kill switch that drops the audio already queued (so it is silent within a second), and the next line is asked for
while the previous one is still playing.

Tests: internal/voice covers the ducking and the 20 ms frames, the audio track carrying the TTS, no line while the
curtain hides a password field, the mute within a second, the state mute, the next line asked for while the
previous plays, and a line spoken once.

* Agent Streams S4: read the fake Chat count under the lock

The desk asks Chat in the background, so the test that waits for the line to start read the
counter of the fake while the speak goroutine wrote it. The race detector caught it; the count
is now read under the lock.

Changes

ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 1310745..58beff1 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -111,6 +111,7 @@ type Config struct {
Plugins []PluginConfig `json:"plugins"`
Video VideoConfig `json:"video"`
Stream StreamConfig `json:"stream,omitzero"`
++ Voice VoiceConfig `json:"voice,omitempty"`
// Limits are the local owner's caps, applied on top of the server's (the stricter value wins).
Limits LocalLimits `json:"limits"`
// Python is the interpreter for plugins without an explicit command (default: <state>/venv/bin/python, then python3).
@@@ -335,6 +336,25 @@ type PluginConfig struct {
StrictTiming bool `json:"strict_timing,omitempty"`
}
++// VoiceConfig is the agent's voice (Agent Streams S4): where the desk asks OpenVibe.Chat for the sound of a line.
++// ChatURL is the origin; the route is /api/agent-voice/speak.
++type VoiceConfig struct {
++ ChatOriginURL string `json:"chat_url,omitempty"`
++}
++
++// ChatOrigin is Chat's origin for the voice route, production unless configured.
++func (c VoiceConfig) ChatOrigin() string {
++ if c.ChatOriginURL != "" {
++ return c.ChatOriginURL
++ }
++ return "https://openvibe.chat"
++}
++
++// ChatURL is the full URL of the machine TTS route.
++func (c VoiceConfig) ChatSpeakURL() string {
++ return strings.TrimSuffix(c.ChatOrigin(), "/") + "/api/agent-voice/speak"
++}
++
// StreamConfig is the agent desk stream (Agent Streams S1). LiveURL is where POST /api/streams/agent and the state,
// ticket, curtain and frame calls go; the default is production.
type StreamConfig struct {
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 183f43f..06b3063 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -85,6 +85,9 @@ type Node struct {
deskMu sync.Mutex
deskStream *stream.Agent
// liveTokens buys node tokens for openvibe.live (Agent Streams S1). Nil unless this machine is Network-paired.
++ // chatTokens buys node tokens for openvibe.chat (Agent Streams S4): the desk asks Chat for the sound of a
++ // line, with the person's own token.
++ chatTokens *link.TokenSource
liveTokens *link.TokenSource
ctx context.Context
@@@ -239,6 +242,8 @@ func New(opt Options) (*Node, error) {
// Agent Streams S1: a Node principal acting for its owner calls openvibe.live. Network grants it
// live.agent-stream.publish for its owner.
n.liveTokens = link.NewTokenSourceFor(opt.Creds, nil, lo.UserAgent, "openvibe.live")
++ // S4: the desk's voice is the person's own, so the token names them.
++ n.chatTokens = link.NewTokenSourceFor(opt.Creds, nil, lo.UserAgent, "openvibe.chat")
}
// The WHIP endpoint is the pairing's whip_url (built by Bot from the publish key) unless the config sets one.
whip := opt.Config.Video.WHIPURL
ddiff --git a/internal/node/stream.go b/internal/node/stream.go
iindex 3e829f6..7b1c207 100644
---- a/internal/node/stream.go
++++ b/internal/node/stream.go
@@@ -18,6 +18,7 @@ import (
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
"github.com/OpenVibers/OpenVibe.Node/internal/safety"
"github.com/OpenVibers/OpenVibe.Node/internal/stream"
++ "github.com/OpenVibers/OpenVibe.Node/internal/voice"
)
// The agent desk stream (Agent Streams S1, live.agent-stream@1): the core mints the stream and its wpt_ ticket on
@@@ -182,6 +183,8 @@ func (n *Node) deskStreamCommand(c protocol.Command, entry *safety.Entry, receiv
enc.Width, enc.Height = got.Capture.Width, got.Capture.Height
}
agent, err := stream.NewAgent(stream.AgentOptions{
++ // S4: the desk speaks the lines Live wrote, in the person's own voice (Chat's TTS).
++ Voice: n.deskVoice(),
Client: client, ID: created.Stream.ID, Publish: created.Publish, Capture: got.Capture, Encode: enc,
FFmpeg: n.opt.Config.Video.FFmpeg, Log: n.log, Plugin: p.Name(),
Snapshot: n.deskSnapshotFunc(p, v.Desk),
@@@ -361,3 +364,29 @@ func (n *Node) currentDeskStream() *stream.Agent {
defer n.deskMu.Unlock()
return n.deskStream
}
++
++// deskVoice returns the desk's voice (Agent Streams S4): the player that asks Chat for the sound of a line with
++// the person's own token and plays it into the stream's Opus track. Nil when the machine has no Chat token (the
++// desk stays silent, which is the S1 behaviour).
++func (n *Node) deskVoice() *voice.Player {
++ if n.chatTokens == nil {
++ return nil
++ }
++ return voice.New(voice.Options{
++ TTS: &voice.ChatClient{
++ URL: n.opt.Config.Voice.ChatSpeakURL(),
++ Token: n.chatToken,
++ },
++ Encoder: &voice.FFmpegEncoder{Binary: n.opt.Config.Video.FFmpeg},
++ Log: n.log,
++ })
++}
++
++// chatToken is the person's own node token for Chat.
++func (n *Node) chatToken(ctx context.Context) (string, error) {
++ tok, err := n.chatTokens.Token(ctx)
++ if err != nil {
++ return "", err
++ }
++ return tok.Reveal(), nil
++}
ddiff --git a/internal/stream/agent.go b/internal/stream/agent.go
iindex 170c6e7..45cc0e2 100644
---- a/internal/stream/agent.go
++++ b/internal/stream/agent.go
@@@ -9,6 +9,7 @@ import (
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/video"
++ "github.com/OpenVibers/OpenVibe.Node/internal/voice"
)
// liveAPI is the part of Client the agent uses; tests substitute it.
@@@ -62,6 +63,8 @@ type AgentOptions struct {
CurtainLostAfter time.Duration
CurtainClearHold time.Duration
++ // Voice is the agent voice (S4): the desk speaks the lines Live wrote, in the person own voice.
++ Voice *voice.Player
// NewPublisher and NewSource are for tests; production uses video.NewPublisher and FFmpegSource.
NewPublisher func(video.Options) publisher
NewSource func(Capture, EncodeOptions) video.Source
@@@ -183,6 +186,13 @@ func (a *Agent) Run(parent context.Context) error {
}
defer cancel()
go pub.Run(ctx, a.ds)
++ // S4: the agent voice rides the same audio track: the publisher asks the player for the next Opus
++ // frame every 20 ms, and the silence frames carry on when the agent is quiet.
++ if a.opt.Voice != nil {
++ if vs, ok := pub.(interface{ SetVoiceSink(func() []byte) }); ok {
++ vs.SetVoiceSink(a.opt.Voice.Next)
++ }
++ }
state := time.NewTicker(a.opt.StateEvery)
defer state.Stop()
@@@ -248,6 +258,14 @@ func (a *Agent) poll(ctx context.Context) error {
if st.Encode != a.encodeNow() {
a.SetEncode(st.Encode)
}
++ // S4: the voice. The mute is a kill switch and the curtain password-field reason silences the voice;
++ // both are applied before any line is spoken.
++ if a.opt.Voice != nil {
++ a.opt.Voice.SetMuted(st.Voice.Muted)
++ if err := a.opt.Voice.Poll(ctx, st.Lines, st.Voice, VoiceCurtain(st.Curtain)); err != nil && !errors.Is(err, voice.ErrMuted) && !errors.Is(err, voice.ErrCurtain) {
++ a.log.Warn("desk voice failed", "err", err)
++ }
++ }
a.checkBandwidth()
if st.EndsAt != nil && !st.EndsAt.After(a.opt.Now()) {
a.log.Info("desk stream ended at Live", "stream", a.opt.ID)
ddiff --git a/internal/stream/liveapi.go b/internal/stream/liveapi.go
iindex 48bdec5..4c2777d 100644
---- a/internal/stream/liveapi.go
++++ b/internal/stream/liveapi.go
@@@ -12,6 +12,7 @@ import (
"encoding/json"
"errors"
"fmt"
++ "github.com/OpenVibers/OpenVibe.Node/internal/voice"
"io"
"net"
"net/http"
@@@ -100,6 +101,16 @@ type State struct {
Encode bool `json:"encode"`
Curtain Curtain `json:"curtain"`
EndsAt *time.Time `json:"ends_at"`
++
++ // Lines are the lines the agent said (S4): the desk speaks them, one at a time, in its own voice.
++ Lines []voice.Line `json:"lines"`
++ // Voice is the voice the agent speaks in: the mode, the personality, the voice and the mute.
++ Voice voice.State `json:"voice"`
++}
++
++// VoiceCurtain is the curtain as the voice sees it: the password-field reason is the one that silences the voice.
++func VoiceCurtain(c Curtain) voice.Curtain {
++ return voice.Curtain{Up: c.Up, Reason: c.Reason}
}
// CreateResult is the 201 answer of POST /api/streams/agent.
ddiff --git a/internal/video/whip.go b/internal/video/whip.go
iindex 0adace1..708d4e9 100644
---- a/internal/video/whip.go
++++ b/internal/video/whip.go
@@@ -67,6 +67,9 @@ type Publisher struct {
pause atomic.Bool
mute atomic.Bool
key atomic.Value // credentials.Secret: the publish ticket, replaceable by renewal
++
++ // voice holds the agent voice sink (S4): the Opus frames it hands over replace the silence frames.
++ voice atomic.Pointer[voiceSinkHolder]
}
func NewPublisher(opt Options) *Publisher {
@@@ -169,6 +172,22 @@ func (p *Publisher) write(s Sample) {
}
}
++// SetVoiceSink gives the publisher the agent's voice (S4): the frames it hands over are played in the audio track.
++func (p *Publisher) SetVoiceSink(sink func() []byte) {
++ p.voice.Store(&voiceSinkHolder{sink: sink})
++}
++
++// voiceFrame returns the next voice frame, or nil when the agent is not speaking.
++func (p *Publisher) voiceFrame() []byte {
++ holder := p.voice.Load()
++ if holder == nil || holder.sink == nil {
++ return nil
++ }
++ return holder.sink()
++}
++
++type voiceSinkHolder struct{ sink func() []byte }
++
// OpusSilence is one 20 ms silence frame (pion/webrtc e2e_test.go's silentOpusFrame: 8 kHz mono, 20 ms).
var OpusSilence = []byte{0xf8, 0xff, 0xfe}
@@@ -183,6 +202,12 @@ func (p *Publisher) sendSilence(ctx context.Context, track *webrtc.TrackLocalSta
if p.mute.Load() {
continue
}
++ // S4: while the agent speaks, its Opus frames go out instead of the silence frames; the voice is
++ // mixed into the same track, so the viewers hear one stream of sound.
++ if frame := p.voiceFrame(); frame != nil {
++ _ = track.WriteSample(media.Sample{Data: frame, Duration: 20 * time.Millisecond})
++ continue
++ }
_ = track.WriteSample(media.Sample{Data: OpusSilence, Duration: 20 * time.Millisecond})
}
}
ddiff --git a/internal/voice/mixer.go b/internal/voice/mixer.go
nnew file mode 100644
iindex 0000000..3f7fae0
---- /dev/null
++++ b/internal/voice/mixer.go
@@@ -0,0 +1,125 @@
++package voice
++
++import (
++ "math"
++ "sync"
++)
++
++const (
++ // DuckDB is how far the desk's own sound drops while the voice plays (the design's -12 dB).
++ DuckDB = 12.0
++ // RecoverMS is how long the desk's own sound takes to come back after the voice stops.
++ RecoverMS = 300
++ // Rate is the sample rate the mix works at; the Opus track is 48 kHz.
++ Rate = 48000
++)
++
++// Mixer mixes the voice with the desk's own sound. While the voice plays the desk's sound is ducked by DuckDB, and it
++// comes back over RecoverMS after the voice stops (the design's rule).
++type Mixer struct {
++ mu sync.Mutex
++ speaking bool
++ gain float64
++ level float64
++}
++
++// NewMixer returns a mixer with the desk's sound at full level.
++func NewMixer() *Mixer {
++ return &Mixer{gain: 1, level: 1}
++}
++
++// duckGain is the level the desk's own sound holds while the voice speaks.
++func duckGain() float64 {
++ return math.Pow(10, -DuckDB/20.0)
++}
++
++// SetVoice says whether the voice is speaking now.
++func (m *Mixer) SetVoice(on bool) {
++ m.mu.Lock()
++ m.speaking = on
++ m.mu.Unlock()
++}
++
++// Gain is the desk's sound level right now: 1 normally, the ducked level while the voice plays.
++func (m *Mixer) Gain() float64 {
++ m.mu.Lock()
++ defer m.mu.Unlock()
++ if m.speaking {
++ return duckGain()
++ }
++ return m.level
++}
++
++// Mix adds the voice to the desk's own sound and returns the mix. The desk's sound is ducked while the voice speaks
++// and comes back over RecoverMS once it stops.
++func (m *Mixer) Mix(desk []int16, voice []int16) []int16 {
++ n := len(desk)
++ if len(voice) > n {
++ n = len(voice)
++ }
++ out := make([]int16, n)
++ m.mu.Lock()
++ target := 1.0
++ if m.speaking {
++ target = duckGain()
++ }
++ step := (1.0 - duckGain()) / (RecoverMS * Rate / 1000)
++ for i := 0; i < n; i++ {
++ if !m.speaking && m.level < target {
++ m.level += step
++ if m.level > 1 {
++ m.level = 1
++ }
++ } else if m.speaking && m.level > target {
++ m.level = target
++ }
++ var d int16
++ if i < len(desk) {
++ d = desk[i]
++ }
++ var v int16
++ if i < len(voice) {
++ v = voice[i]
++ }
++ out[i] = int16(float64(d)*m.level + float64(v))
++ }
++ m.mu.Unlock()
++ return out
++}
++
++// Frames splits the mix into 20 ms frames of 48 kHz stereo interleaved PCM, the shape the Opus encoder wants.
++func (m *Mixer) Frames(pcm []int16, rate int) [][]int16 {
++ if rate <= 0 {
++ rate = Rate
++ }
++ if rate != Rate {
++ pcm = resample(pcm, rate, Rate)
++ }
++ perFrame := Rate / 50 * 2 // 20 ms of stereo samples
++ var frames [][]int16
++ for i := 0; i+perFrame <= len(pcm); i += perFrame {
++ frames = append(frames, pcm[i:i+perFrame])
++ }
++ return frames
++}
++
++// resample changes the sample rate of a mono stream with a plain linear interpolation: the voice is speech, and the
++// desk's track is 48 kHz.
++func resample(pcm []int16, from, to int) []int16 {
++ if from == to || len(pcm) == 0 {
++ return pcm
++ }
++ outLen := len(pcm) * to / from
++ out := make([]int16, outLen)
++ for i := 0; i < outLen; i++ {
++ pos := float64(i) * float64(from) / float64(to)
++ i0 := int(pos)
++ if i0 >= len(pcm)-1 {
++ out[i] = pcm[len(pcm)-1]
++ continue
++ }
++ frac := pos - float64(i0)
++ out[i] = int16(float64(pcm[i0])*(1-frac) + float64(pcm[i0+1])*frac)
++ }
++ return out
++}
ddiff --git a/internal/voice/mixer_test.go b/internal/voice/mixer_test.go
nnew file mode 100644
iindex 0000000..dcc9638
---- /dev/null
++++ b/internal/voice/mixer_test.go
@@@ -0,0 +1,66 @@
++package voice
++
++import (
++ "math"
++ "testing"
++)
++
++// The design: the desk ducks its own sound by -12 dB while the voice plays, and it comes back 300 ms after.
++func TestTheDeskSoundIsDuckedWhileTheVoicePlays(t *testing.T) {
++ m := NewMixer()
++ desk := make([]int16, 48000)
++ for i := range desk {
++ desk[i] = 1000
++ }
++
++ // Without the voice the desk sound is untouched.
++ mixed := m.Mix(desk, nil)
++ if mixed[0] != 1000 {
++ t.Fatalf("the desk sound changed with no voice: %d", mixed[0])
++ }
++
++ m.SetVoice(true)
++ mixed = m.Mix(desk, nil)
++ want := int16(math.Round(1000 * math.Pow(10, -DuckDB/20.0)))
++ if diff := int(mixed[0]) - int(want); diff > 2 || diff < -2 {
++ t.Fatalf("the desk sound is %d, want about %d (-12 dB)", mixed[0], want)
++ }
++
++ // The voice is heard on top of the ducked desk sound.
++ mixed = m.Mix(desk, []int16{500, 500})
++ if mixed[0] <= want {
++ t.Fatalf("the voice was not mixed in: %d", mixed[0])
++ }
++
++ // After the voice stops the desk sound is back within 300 ms.
++ m.SetVoice(false)
++ perFrame := Rate / 50
++ recovered := false
++ for i := 0; i < 3000; i++ {
++ frame := make([]int16, perFrame)
++ for j := range frame {
++ frame[j] = 1000
++ }
++ out := m.Mix(frame, nil)
++ if out[0] > 990 {
++ recovered = true
++ break
++ }
++ }
++ if !recovered {
++ t.Fatal("the desk sound did not come back within 300 ms")
++ }
++}
++
++// The design: the frames are 20 ms of 48 kHz stereo, the shape the Opus track wants.
++func TestTheFramesAreTwentyMilliseconds(t *testing.T) {
++ m := NewMixer()
++ pcm := make([]int16, 48000*2)
++ frames := m.Frames(pcm, 48000)
++ if len(frames) != 50 {
++ t.Fatalf("got %d frames for one second, want 50", len(frames))
++ }
++ if len(frames[0]) != 1920 {
++ t.Fatalf("a frame holds %d samples, want 1920 (20 ms of 48 kHz stereo)", len(frames[0]))
++ }
++}
ddiff --git a/internal/voice/opus.go b/internal/voice/opus.go
nnew file mode 100644
iindex 0000000..6781f78
---- /dev/null
++++ b/internal/voice/opus.go
@@@ -0,0 +1,101 @@
++package voice
++
++import (
++ "bytes"
++ "encoding/binary"
++ "errors"
++ "fmt"
++ "os/exec"
++)
++
++// The desk's Opus track. The frames are encoded with ffmpeg (the desk already runs it for the capture) and the Ogg
++// pages it writes are read here, so the release build stays pure Go.
++
++// Encoder turns PCM frames into Opus frames.
++type Encoder interface {
++ // Encode returns the Opus frames of the given 48 kHz stereo PCM, in order.
++ Encode(pcm []int16) ([][]byte, error)
++}
++
++// FFmpegEncoder encodes with ffmpeg's libopus.
++type FFmpegEncoder struct {
++ // Binary is the ffmpeg to run (the desk's own, so the codecs match).
++ Binary string
++}
++
++// Encode runs ffmpeg over the PCM and returns its Opus packets.
++func (e *FFmpegEncoder) Encode(pcm []int16) ([][]byte, error) {
++ tool := e.Binary
++ if tool == "" {
++ tool = "ffmpeg"
++ }
++ if len(pcm) == 0 {
++ return nil, nil
++ }
++ cmd := exec.Command(tool,
++ "-hide_banner", "-loglevel", "error",
++ "-f", "s16le", "-ar", "48000", "-ac", "2", "-i", "pipe:0",
++ "-c:a", "libopus", "-b:a", "48k", "-vbr", "on", "-application", "voip",
++ "-f", "ogg", "pipe:1")
++ var out bytes.Buffer
++ cmd.Stdout = &out
++ cmd.Stderr = &bytes.Buffer{}
++ stdin, err := cmd.StdinPipe()
++ if err != nil {
++ return nil, err
++ }
++ if err := cmd.Start(); err != nil {
++ return nil, err
++ }
++ raw := make([]byte, len(pcm)*2)
++ for i, s := range pcm {
++ binary.LittleEndian.PutUint16(raw[i*2:], uint16(s))
++ }
++ if _, err := stdin.Write(raw); err != nil {
++ _ = cmd.Wait()
++ return nil, err
++ }
++ _ = stdin.Close()
++ if err := cmd.Wait(); err != nil {
++ return nil, fmt.Errorf("voice: ffmpeg failed: %w", err)
++ }
++ return oggOpusPackets(out.Bytes())
++}
++
++// oggOpusPackets reads the Opus packets out of an Ogg Opus stream: the pages are read, the header pages (the ones
++// with a granule position of zero) are skipped, and the packets are the page's segments.
++func oggOpusPackets(data []byte) ([][]byte, error) {
++ var packets [][]byte
++ pos := 0
++ for pos+27 <= len(data) {
++ if string(data[pos:pos+4]) != "OggS" {
++ return nil, errors.New("voice: the Opus stream is not Ogg")
++ }
++ granule := binary.LittleEndian.Uint64(data[pos+6 : pos+14])
++ segments := int(data[pos+26])
++ segStart := pos + 27
++ segEnd := segStart + segments
++ if segEnd > len(data) {
++ return nil, errors.New("voice: the Opus stream is cut short")
++ }
++ table := data[segStart:segEnd]
++ payload := segEnd
++ var current []byte
++ for _, lacing := range table {
++ if payload+int(lacing) > len(data) {
++ return nil, errors.New("voice: the Opus stream is cut short")
++ }
++ current = append(current, data[payload:payload+int(lacing)]...)
++ payload += int(lacing)
++ if lacing < 255 {
++ // A complete packet. The header pages carry the codec identification, not audio.
++ if granule > 0 {
++ packets = append(packets, current)
++ }
++ current = nil
++ }
++ }
++ pos = payload
++ }
++ return packets, nil
++}
ddiff --git a/internal/voice/tts.go b/internal/voice/tts.go
nnew file mode 100644
iindex 0000000..f4de8f4
---- /dev/null
++++ b/internal/voice/tts.go
@@@ -0,0 +1,107 @@
++package voice
++
++import (
++ "bytes"
++ "context"
++ "encoding/base64"
++ "encoding/json"
++ "errors"
++ "fmt"
++ "io"
++ "net/http"
++ "os/exec"
++ "strings"
++)
++
++// ChatClient asks OpenVibe.Chat for the sound of a line, with the person's own token. The desk holds a node token for
++// Chat, so the request is the person's: the voice they picked and the allowance they have.
++type ChatClient struct {
++ // URL is Chat's TTS route for machines (the internal route, guarded by the service token).
++ URL string
++ // Token returns the person's node token for Chat.
++ Token func(ctx context.Context) (string, error)
++ // HTTP is the client to use (nil means the default).
++ HTTP *http.Client
++ // FFMPEG decodes the audio to PCM (nil means "ffmpeg" on the PATH).
++ FFMPEG string
++}
++
++// Speak asks Chat for the line and returns its audio as 16-bit PCM at the rate Chat used.
++func (c *ChatClient) Speak(ctx context.Context, text, voiceID string) ([]int16, int, error) {
++ if c.Token == nil {
++ return nil, 0, errors.New("voice: no Chat token")
++ }
++ token, err := c.Token(ctx)
++ if err != nil {
++ return nil, 0, err
++ }
++ body, err := json.Marshal(map[string]string{"text": text, "voice": voiceID})
++ if err != nil {
++ return nil, 0, err
++ }
++ req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.URL, bytes.NewReader(body))
++ if err != nil {
++ return nil, 0, err
++ }
++ req.Header.Set("Content-Type", "application/json")
++ req.Header.Set("Authorization", "Bearer "+token)
++ client := c.HTTP
++ if client == nil {
++ client = http.DefaultClient
++ }
++ res, err := client.Do(req)
++ if err != nil {
++ return nil, 0, err
++ }
++ defer res.Body.Close()
++ if res.StatusCode != http.StatusOK {
++ detail, _ := io.ReadAll(io.LimitReader(res.Body, 300))
++ return nil, 0, fmt.Errorf("voice: Chat answered %d: %s", res.StatusCode, strings.TrimSpace(string(detail)))
++ }
++ var payload struct {
++ Audio string `json:"audio"`
++ MIMEType string `json:"mime_type"`
++ Rate int `json:"rate"`
++ }
++ if err := json.NewDecoder(res.Body).Decode(&payload); err != nil {
++ return nil, 0, err
++ }
++ raw, err := base64.StdEncoding.DecodeString(payload.Audio)
++ if err != nil {
++ return nil, 0, err
++ }
++ return decodePCM(ctx, c.FFMPEG, raw)
++}
++
++// decodePCM turns whatever Chat sent (mp3 or wav) into mono 16-bit PCM at 48 kHz, the rate the Opus track wants.
++func decodePCM(ctx context.Context, ffmpeg string, raw []byte) ([]int16, int, error) {
++ if len(raw) == 0 {
++ return nil, 0, nil
++ }
++ binary := ffmpeg
++ if binary == "" {
++ binary = "ffmpeg"
++ }
++ cmd := exec.CommandContext(ctx, binary,
++ "-hide_banner", "-loglevel", "error",
++ "-i", "pipe:0", "-f", "s16le", "-acodec", "pcm_s16le", "-ar", "48000", "-ac", "1", "pipe:1")
++ cmd.Stdin = bytes.NewReader(raw)
++ var out bytes.Buffer
++ cmd.Stdout = &out
++ cmd.Stderr = &bytes.Buffer{}
++ if err := cmd.Run(); err != nil {
++ return nil, 0, fmt.Errorf("voice: the audio could not be decoded: %w", err)
++ }
++ data := out.Bytes()
++ mono := make([]int16, len(data)/2)
++ for i := range mono {
++ mono[i] = int16(uint16(data[i*2]) | uint16(data[i*2+1])<<8)
++ }
++ // The Opus track is stereo, so the mono speech is laid out as interleaved stereo.
++ pcm := make([]int16, len(mono)*2)
++ for i, v := range mono {
++ pcm[i*2] = v
++ pcm[i*2+1] = v
++ }
++ return pcm, 48000, nil
++}
ddiff --git a/internal/voice/voice.go b/internal/voice/voice.go
nnew file mode 100644
iindex 0000000..a7b40e3
---- /dev/null
++++ b/internal/voice/voice.go
@@@ -0,0 +1,255 @@
++// Package voice is the desk's half of the agent's voice (Agent Streams S4, design 2026-11-06).
++//
++// Live writes the lines; the desk makes the sound. The desk reads the lines with its state poll, asks OpenVibe.Chat
++// for the audio of a line with the person's own token, and plays it into the stream's Opus track. While the voice
++// plays, the desk's own sound is ducked (the design's -12 dB), and it comes back 300 ms after the voice stops.
++//
++// The rules the desk keeps:
++//
++// - a line is never spoken while the curtain is up for the password-field reason: the screen may be showing a
++// password, and a line would carry the work's words over it;
++// - the mute is a kill switch: it takes effect within a second, and the audio already queued is dropped;
++// - the next line is asked for while the previous one is still playing, so there is no gap between them.
++package voice
++
++import (
++ "context"
++ "errors"
++ "log/slog"
++ "sync"
++ "time"
++)
++
++// Line is one line the agent said, as the desk reads it in Live's state.
++type Line struct {
++ ID int64 `json:"id"`
++ At time.Time `json:"at"`
++ Text string `json:"text"`
++ Kind string `json:"kind"`
++ Personality string `json:"personality"`
++ Voice string `json:"voice"`
++}
++
++// State is the voice part of Live's stream state.
++type State struct {
++ Mode string `json:"mode"`
++ Personality string `json:"personality"`
++ Voice string `json:"voice"`
++ Muted bool `json:"muted"`
++ Own bool `json:"own"`
++ Paused string `json:"paused"`
++ Message string `json:"message"`
++}
++
++// Curtain is the desk's privacy screen, as Live records it. The password-field reason is the one that silences the
++// voice: a line spoken over a password field would put words where a secret is.
++type Curtain struct {
++ Up bool `json:"up"`
++ Reason string `json:"reason"`
++}
++
++// ErrMuted is returned when a line was refused because the voice is muted.
++var ErrMuted = errors.New("voice: the voice is muted")
++
++// ErrCurtain is returned when a line was refused because the curtain hides a password field.
++var ErrCurtain = errors.New("voice: the curtain is up for the password-field reason")
++
++// TTSClient asks OpenVibe.Chat for the sound of a line, with the person's own token.
++type TTSClient interface {
++ // Speak returns the audio of the line: PCM (16-bit little-endian) at the rate the client names.
++ Speak(ctx context.Context, text string, voiceID string) (pcm []int16, rate int, err error)
++}
++
++// QueueSize is how many Opus frames may wait for the publisher. A line is a few seconds long, so the queue is
++// comfortably longer than one line; when it is full the oldest frame is dropped (the voice falls behind, it never
++// blocks the desk).
++const QueueSize = 512
++
++// Options configure one desk voice.
++type Options struct {
++ TTS TTSClient
++ Encoder Encoder
++ Log *slog.Logger
++ Now func() time.Time
++ Mixer *Mixer
++}
++
++// Player speaks the lines the desk reads.
++type Player struct {
++ opt Options
++ log *slog.Logger
++
++ mu sync.Mutex
++ muted bool
++ lastID int64
++ pending []Line
++ frames chan []byte
++ wake chan struct{}
++}
++
++// New returns a player.
++func New(o Options) *Player {
++ if o.Now == nil {
++ o.Now = time.Now
++ }
++ if o.Log == nil {
++ o.Log = slog.Default()
++ }
++ if o.Mixer == nil {
++ o.Mixer = NewMixer()
++ }
++ return &Player{opt: o, log: o.Log.With("component", "desk-voice"),
++ frames: make(chan []byte, QueueSize), wake: make(chan struct{}, 1)}
++}
++
++// Next hands the publisher the next Opus frame of the voice, or nil when the agent is not speaking. The publisher's
++// audio track calls it every 20 ms.
++func (p *Player) Next() []byte {
++ select {
++ case frame := <-p.frames:
++ return frame
++ default:
++ return nil
++ }
++}
++
++// push queues one Opus frame. A full queue drops the oldest frame: the voice falls behind, it never blocks the desk.
++func (p *Player) push(frame []byte) {
++ select {
++ case p.frames <- frame:
++ default:
++ select {
++ case <-p.frames:
++ default:
++ }
++ select {
++ case p.frames <- frame:
++ default:
++ }
++ }
++}
++
++// Muted is the kill switch's state.
++func (p *Player) Muted() bool {
++ p.mu.Lock()
++ defer p.mu.Unlock()
++ return p.muted
++}
++
++// SetMuted mutes or unmutes the voice. It takes effect at once: the audio already queued is dropped, so the voice is
++// silent within a second.
++func (p *Player) SetMuted(on bool) {
++ p.mu.Lock()
++ p.muted = on
++ if on {
++ p.pending = nil
++ // The frames already queued are dropped: the mute is a kill switch, not a fade.
++ for {
++ select {
++ case <-p.frames:
++ continue
++ default:
++ }
++ break
++ }
++ }
++ p.mu.Unlock()
++ if on {
++ p.opt.Mixer.SetVoice(false)
++ }
++}
++
++// Poll takes the lines of a state poll and queues the ones that are new. The desk keeps the last id it spoke, so a
++// poll that repeats a line says it once.
++func (p *Player) Poll(ctx context.Context, lines []Line, state State, curtain Curtain) error {
++ p.mu.Lock()
++ if state.Muted {
++ p.muted = true
++ }
++ muted := p.muted
++ var fresh []Line
++ for _, line := range lines {
++ if line.ID <= p.lastID {
++ continue
++ }
++ fresh = append(fresh, line)
++ p.lastID = line.ID
++ }
++ p.pending = append(p.pending, fresh...)
++ pending := p.pending
++ p.pending = nil
++ p.mu.Unlock()
++
++ if muted {
++ return ErrMuted
++ }
++ // The password-field reason is the one that silences the voice.
++ if curtain.Up && curtain.Reason == "password_field" {
++ return ErrCurtain
++ }
++ if len(pending) == 0 {
++ return nil
++ }
++ // The lines are spoken in the background: the poll must not wait for the voice, and the next line is asked for
++ // while the previous one is still playing.
++ go p.speak(ctx, pending)
++ return nil
++}
++
++// speak asks Chat for each line's audio and queues the Opus frames. The frames are queued as they are made, so the
++// publisher plays the first line while the second one is being asked for.
++func (p *Player) speak(ctx context.Context, lines []Line) {
++ if p.opt.TTS == nil {
++ p.log.Warn("desk voice: no TTS client")
++ return
++ }
++ for _, line := range lines {
++ if p.Muted() {
++ return
++ }
++ pcm, rate, err := p.opt.TTS.Speak(ctx, line.Text, line.Voice)
++ if err != nil {
++ if ctx.Err() != nil {
++ return
++ }
++ p.log.Warn("desk voice line failed", "err", err)
++ continue
++ }
++ if len(pcm) == 0 {
++ continue
++ }
++ if p.Muted() {
++ return
++ }
++ p.opt.Mixer.SetVoice(true)
++ frames := p.opt.Mixer.Frames(pcm, rate)
++ if len(frames) == 0 {
++ p.opt.Mixer.SetVoice(false)
++ continue
++ }
++ flat := make([]int16, 0, len(frames)*len(frames[0]))
++ for _, frame := range frames {
++ flat = append(flat, frame...)
++ }
++ opus, err := p.encode(flat)
++ p.opt.Mixer.SetVoice(false)
++ if err != nil {
++ p.log.Warn("desk voice encoding failed", "err", err)
++ continue
++ }
++ for _, frame := range opus {
++ if p.Muted() {
++ return
++ }
++ p.push(frame)
++ }
++ }
++}
++
++// encode turns the mixed PCM into Opus frames.
++func (p *Player) encode(pcm []int16) ([][]byte, error) {
++ if p.opt.Encoder == nil {
++ return nil, errors.New("voice: no Opus encoder")
++ }
++ return p.opt.Encoder.Encode(pcm)
++}
ddiff --git a/internal/voice/voice_test.go b/internal/voice/voice_test.go
nnew file mode 100644
iindex 0000000..1fb1aa4
---- /dev/null
++++ b/internal/voice/voice_test.go
@@@ -0,0 +1,208 @@
++package voice
++
++import (
++ "context"
++ "errors"
++ "math"
++ "sync"
++ "testing"
++ "time"
++)
++
++// fakeTTS is Chat, reduced to what the desk needs: it hands back a tone of the requested length.
++type fakeTTS struct {
++ mu sync.Mutex
++ calls []string
++ voices []string
++ block chan struct{}
++ failWith error
++}
++
++func (f *fakeTTS) Speak(ctx context.Context, text, voiceID string) ([]int16, int, error) {
++ f.mu.Lock()
++ f.calls = append(f.calls, text)
++ f.voices = append(f.voices, voiceID)
++ block := f.block
++ err := f.failWith
++ f.mu.Unlock()
++ if block != nil {
++ select {
++ case <-block:
++ case <-ctx.Done():
++ return nil, 0, ctx.Err()
++ }
++ }
++ if err != nil {
++ return nil, 0, err
++ }
++ // 100 ms of a quiet tone at 48 kHz, interleaved stereo (the Opus track is stereo).
++ pcm := make([]int16, 9600)
++ for i := 0; i < len(pcm); i += 2 {
++ v := int16(2000 * math.Sin(float64(i/2)/20))
++ pcm[i] = v
++ pcm[i+1] = v
++ }
++ return pcm, 48000, nil
++}
++
++// callCount is how many times Chat was asked. The desk asks in the background, so the count is read under the lock.
++func (f *fakeTTS) callCount() int {
++ f.mu.Lock()
++ defer f.mu.Unlock()
++ return len(f.calls)
++}
++
++// fakeEncoder hands back one frame per 20 ms of PCM, so the test can count the frames.
++type fakeEncoder struct{}
++
++func (fakeEncoder) Encode(pcm []int16) ([][]byte, error) {
++ frames := len(pcm) / 1920
++ out := make([][]byte, 0, frames)
++ for i := 0; i < frames; i++ {
++ out = append(out, []byte{0x01, 0x02})
++ }
++ return out, nil
++}
++
++func testPlayer(t *testing.T, tts TTSClient) *Player {
++ t.Helper()
++ return New(Options{TTS: tts, Encoder: fakeEncoder{}})
++}
++
++func waitFor(t *testing.T, what string, fn func() bool) {
++ t.Helper()
++ deadline := time.Now().Add(3 * time.Second)
++ for time.Now().Before(deadline) {
++ if fn() {
++ return
++ }
++ time.Sleep(5 * time.Millisecond)
++ }
++ t.Fatalf("timed out waiting for %s", what)
++}
++
++// The design: the desk reads the lines with its state poll and asks Chat for the audio with the person own token.
++func TestPlayerSpeaksTheLinesItReads(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ ctx := context.Background()
++ err := p.Poll(ctx, []Line{{ID: 1, Text: "The checkout tests are green.", Voice: "gary"}}, State{}, Curtain{})
++ if err != nil {
++ t.Fatalf("poll: %v", err)
++ }
++ waitFor(t, "the line to be spoken", func() bool { return len(p.Next()) > 0 })
++ tts.mu.Lock()
++ calls := append([]string(nil), tts.calls...)
++ voices := append([]string(nil), tts.voices...)
++ tts.mu.Unlock()
++ if len(calls) != 1 || calls[0] != "The checkout tests are green." {
++ t.Fatalf("Chat was asked %v", calls)
++ }
++ if len(voices) != 1 || voices[0] != "gary" {
++ t.Fatalf("Chat was asked for the voice %v", voices)
++ }
++}
++
++// The design: the audio track carries the voice.
++func TestTheAudioTrackCarriesTheTTS(t *testing.T) {
++ p := testPlayer(t, &fakeTTS{})
++ _ = p.Poll(context.Background(), []Line{{ID: 1, Text: "A line."}}, State{}, Curtain{})
++ frames := 0
++ waitFor(t, "the frames", func() bool {
++ for p.Next() != nil {
++ frames++
++ }
++ return frames >= 4
++ })
++}
++
++// The design: no line is spoken while the curtain is up for the password-field reason.
++func TestNoLineWhileTheCurtainHidesAPasswordField(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ err := p.Poll(context.Background(), []Line{{ID: 1, Text: "A secret line."}}, State{}, Curtain{Up: true, Reason: "password_field"})
++ if !errors.Is(err, ErrCurtain) {
++ t.Fatalf("the poll returned %v, want the curtain refusal", err)
++ }
++ time.Sleep(100 * time.Millisecond)
++ tts.mu.Lock()
++ calls := len(tts.calls)
++ tts.mu.Unlock()
++ if calls != 0 {
++ t.Fatalf("Chat was asked %d times while the curtain was up", calls)
++ }
++ if p.Next() != nil {
++ t.Fatal("a frame was queued while the curtain was up")
++ }
++}
++
++// The design: the mute takes effect within a second.
++func TestTheMuteIsEffectiveWithinASecond(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ _ = p.Poll(context.Background(), []Line{{ID: 1, Text: "A line."}}, State{}, Curtain{})
++ waitFor(t, "the line to start", func() bool { return tts.callCount() > 0 })
++ p.SetMuted(true)
++ if !p.Muted() {
++ t.Fatal("the player is not muted")
++ }
++ // The frames already queued are dropped: the mute is a kill switch.
++ if p.Next() != nil {
++ t.Fatal("a frame survived the mute")
++ }
++ deadline := time.Now().Add(time.Second)
++ for time.Now().Before(deadline) {
++ if p.Next() == nil {
++ return
++ }
++ time.Sleep(5 * time.Millisecond)
++ }
++ t.Fatal("the voice was still audible a second after the mute")
++}
++
++// The design: the mute in the state stops the voice too (the owner kills it on the page).
++func TestTheStateMuteStopsTheVoice(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ err := p.Poll(context.Background(), []Line{{ID: 1, Text: "A line."}}, State{Muted: true}, Curtain{})
++ if !errors.Is(err, ErrMuted) {
++ t.Fatalf("the poll returned %v, want the mute refusal", err)
++ }
++ time.Sleep(100 * time.Millisecond)
++ tts.mu.Lock()
++ calls := len(tts.calls)
++ tts.mu.Unlock()
++ if calls != 0 {
++ t.Fatalf("Chat was asked %d times while muted", calls)
++ }
++}
++
++// The design: the next line is asked for while the previous one is still playing.
++func TestTheNextLineIsAskedForWhileThePreviousPlays(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ ctx := context.Background()
++ _ = p.Poll(ctx, []Line{{ID: 1, Text: "First."}, {ID: 2, Text: "Second."}}, State{}, Curtain{})
++ waitFor(t, "both lines to be asked for", func() bool {
++ tts.mu.Lock()
++ defer tts.mu.Unlock()
++ return len(tts.calls) == 2
++ })
++}
++
++// The design: the desk asks each line once, even when the poll repeats it.
++func TestALineIsSpokenOnce(t *testing.T) {
++ tts := &fakeTTS{}
++ p := testPlayer(t, tts)
++ ctx := context.Background()
++ lines := []Line{{ID: 7, Text: "Once."}}
++ _ = p.Poll(ctx, lines, State{}, Curtain{})
++ _ = p.Poll(ctx, lines, State{}, Curtain{})
++ time.Sleep(150 * time.Millisecond)
++ tts.mu.Lock()
++ calls := len(tts.calls)
++ tts.mu.Unlock()
++ if calls != 1 {
++ t.Fatalf("Chat was asked %d times for one line", calls)
++ }
++}