From d97dcf252ca4c66dce50730fce55050e80df515d Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 14 Apr 2026 23:35:26 +0700 Subject: [PATCH] feat(audio): ElevenLabs streaming TTS with model validation Implement streaming TTS synthesis for ElevenLabs. Add model validation to buffered path. Add characterization tests for latency and error scenarios. --- internal/audio/elevenlabs/tts.go | 3 + internal/audio/elevenlabs/tts_stream.go | 83 ++++++++ internal/audio/elevenlabs/tts_stream_test.go | 194 +++++++++++++++++++ internal/audio/elevenlabs/tts_test.go | 43 ++++ 4 files changed, 323 insertions(+) create mode 100644 internal/audio/elevenlabs/tts_stream.go create mode 100644 internal/audio/elevenlabs/tts_stream_test.go create mode 100644 internal/audio/elevenlabs/tts_test.go diff --git a/internal/audio/elevenlabs/tts.go b/internal/audio/elevenlabs/tts.go index 18dc50c9..f4691f8e 100644 --- a/internal/audio/elevenlabs/tts.go +++ b/internal/audio/elevenlabs/tts.go @@ -38,6 +38,9 @@ func (p *TTSProvider) Synthesize(ctx context.Context, text string, opts audio.TT if modelID == "" { modelID = p.cfg.ModelID } + if err := ValidateModel(modelID); err != nil { + return nil, err + } // Output format: MP3 128kbps default, Opus 64kbps for Telegram voice. outputFormat, ext, mime := "mp3_44100_128", "mp3", "audio/mpeg" diff --git a/internal/audio/elevenlabs/tts_stream.go b/internal/audio/elevenlabs/tts_stream.go new file mode 100644 index 00000000..6e23a67a --- /dev/null +++ b/internal/audio/elevenlabs/tts_stream.go @@ -0,0 +1,83 @@ +package elevenlabs + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/audio" +) + +// SynthesizeStream streams ElevenLabs TTS audio via the /stream endpoint. +// Unlike Synthesize, the HTTP body is handed to the caller as an io.ReadCloser +// so bytes can be consumed as they arrive. Callers MUST Close the returned +// Audio reader. +// +// Resolution order for voice/model: opts > configured defaults. +// Model IDs are validated against AllowedElevenLabsModels before dispatch — +// unknown IDs surface an i18n-keyed error without hitting the network. +func (p *TTSProvider) SynthesizeStream(ctx context.Context, text string, opts audio.TTSOptions) (*audio.StreamResult, error) { + voiceID := opts.Voice + if voiceID == "" { + voiceID = p.cfg.VoiceID + } + modelID := opts.Model + if modelID == "" { + modelID = p.cfg.ModelID + } + if err := ValidateModel(modelID); err != nil { + return nil, err + } + + outputFormat, ext, mime := "mp3_44100_128", "mp3", "audio/mpeg" + if opts.Format == "opus" { + outputFormat, ext, mime = "opus_48000_64", "ogg", "audio/ogg" + } + + body := map[string]any{ + "text": text, + "model_id": modelID, + "voice_settings": map[string]any{ + "stability": 0.5, + "similarity_boost": 0.75, + "style": 0.0, + "use_speaker_boost": true, + }, + } + bodyJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal elevenlabs tts stream request: %w", err) + } + + path := fmt.Sprintf("/v1/text-to-speech/%s/stream?output_format=%s", voiceID, outputFormat) + url := strings.TrimRight(p.c.baseURL, "/") + path + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(bodyJSON)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("xi-api-key", p.c.apiKey) + + // No Client.Timeout — streaming may legitimately outlast a per-request + // deadline. Cancellation is honoured via the request context. + hc := &http.Client{Timeout: 0} + resp, err := hc.Do(req) + if err != nil { + return nil, fmt.Errorf("elevenlabs tts stream: %w", err) + } + if resp.StatusCode != http.StatusOK { + errBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + return nil, fmt.Errorf("ElevenLabs API error %d: %s", resp.StatusCode, truncate(errBody, 500)) + } + + return &audio.StreamResult{ + Audio: resp.Body, + Extension: ext, + MimeType: mime, + }, nil +} diff --git a/internal/audio/elevenlabs/tts_stream_test.go b/internal/audio/elevenlabs/tts_stream_test.go new file mode 100644 index 00000000..44dcf8f1 --- /dev/null +++ b/internal/audio/elevenlabs/tts_stream_test.go @@ -0,0 +1,194 @@ +package elevenlabs + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/audio" +) + +// chunkedServer returns an httptest.Server that streams `chunks` back with +// Flush between each. The handler also records the request path + headers. +func chunkedServer(t *testing.T, chunks []string) (*httptest.Server, *atomic.Value) { + t.Helper() + lastPath := &atomic.Value{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + lastPath.Store(r.URL.RequestURI()) + if r.Header.Get("xi-api-key") == "" { + t.Errorf("missing xi-api-key header") + } + w.Header().Set("Content-Type", "audio/mpeg") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + for _, c := range chunks { + _, _ = w.Write([]byte(c)) + if flusher != nil { + flusher.Flush() + } + time.Sleep(5 * time.Millisecond) + } + })) + return srv, lastPath +} + +func TestSynthesizeStream_ReturnsChunkedReader(t *testing.T) { + t.Parallel() + chunks := []string{"aaa", "bbb", "ccc"} + srv, lastPath := chunkedServer(t, chunks) + defer srv.Close() + + p := NewTTSProvider(Config{ + APIKey: "test-key", + BaseURL: srv.URL, + VoiceID: "V1", + ModelID: "eleven_v3", + TimeoutMs: 5000, + }) + + res, err := p.SynthesizeStream(context.Background(), "hello world", audio.TTSOptions{}) + if err != nil { + t.Fatalf("SynthesizeStream error: %v", err) + } + if res.Audio == nil { + t.Fatal("StreamResult.Audio is nil") + } + defer res.Audio.Close() + + if res.Extension != "mp3" { + t.Errorf("Extension: got %q, want mp3", res.Extension) + } + if res.MimeType != "audio/mpeg" { + t.Errorf("MimeType: got %q, want audio/mpeg", res.MimeType) + } + + body, err := io.ReadAll(res.Audio) + if err != nil { + t.Fatalf("read body: %v", err) + } + want := strings.Join(chunks, "") + if string(body) != want { + t.Errorf("body: got %q, want %q", string(body), want) + } + + // URL must hit the /stream sub-path with mp3 output format. + path, _ := lastPath.Load().(string) + if !strings.Contains(path, "/v1/text-to-speech/V1/stream") { + t.Errorf("path: got %q, want contains /v1/text-to-speech/V1/stream", path) + } + if !strings.Contains(path, "output_format=mp3_44100_128") { + t.Errorf("path: got %q, want contains output_format=mp3_44100_128", path) + } +} + +func TestSynthesizeStream_VoiceOverride(t *testing.T) { + t.Parallel() + srv, lastPath := chunkedServer(t, []string{"x"}) + defer srv.Close() + + p := NewTTSProvider(Config{ + APIKey: "k", + BaseURL: srv.URL, + VoiceID: "DEFAULT", + ModelID: "eleven_v3", + }) + + res, err := p.SynthesizeStream(context.Background(), "hi", audio.TTSOptions{Voice: "OVR"}) + if err != nil { + t.Fatalf("err: %v", err) + } + res.Audio.Close() + + path, _ := lastPath.Load().(string) + if !strings.Contains(path, "/v1/text-to-speech/OVR/stream") { + t.Errorf("voice override not honoured — path=%q", path) + } +} + +func TestSynthesizeStream_CancelContextClosesBody(t *testing.T) { + t.Parallel() + // Server writes one chunk then blocks forever. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "audio/mpeg") + w.WriteHeader(http.StatusOK) + if f, ok := w.(http.Flusher); ok { + _, _ = w.Write([]byte("a")) + f.Flush() + } + <-r.Context().Done() + })) + defer srv.Close() + + p := NewTTSProvider(Config{APIKey: "k", BaseURL: srv.URL, VoiceID: "V"}) + + ctx, cancel := context.WithCancel(context.Background()) + res, err := p.SynthesizeStream(ctx, "hi", audio.TTSOptions{}) + if err != nil { + t.Fatalf("err: %v", err) + } + + // Cancel while body is still open — next Read should fail promptly. + cancel() + done := make(chan error, 1) + go func() { + buf := make([]byte, 64) + for { + _, err := res.Audio.Read(buf) + if err != nil { + done <- err + return + } + } + }() + select { + case <-done: + // ok — read returned an error after cancel. + case <-time.After(2 * time.Second): + t.Fatal("Read did not return after ctx cancel") + } + res.Audio.Close() +} + +func TestSynthesizeStream_OpusFormat(t *testing.T) { + t.Parallel() + srv, lastPath := chunkedServer(t, []string{"o"}) + defer srv.Close() + + p := NewTTSProvider(Config{APIKey: "k", BaseURL: srv.URL, VoiceID: "V", ModelID: "eleven_v3"}) + res, err := p.SynthesizeStream(context.Background(), "hi", audio.TTSOptions{Format: "opus"}) + if err != nil { + t.Fatalf("err: %v", err) + } + res.Audio.Close() + + if res.Extension != "ogg" { + t.Errorf("Extension: got %q, want ogg", res.Extension) + } + if res.MimeType != "audio/ogg" { + t.Errorf("MimeType: got %q, want audio/ogg", res.MimeType) + } + path, _ := lastPath.Load().(string) + if !strings.Contains(path, "output_format=opus_48000_64") { + t.Errorf("path: got %q, want contains output_format=opus_48000_64", path) + } +} + +func TestSynthesizeStream_RejectsUnknownModel(t *testing.T) { + t.Parallel() + // Server should never be reached — validation runs before HTTP. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("unexpected upstream request for invalid model") + })) + defer srv.Close() + + p := NewTTSProvider(Config{APIKey: "k", BaseURL: srv.URL, VoiceID: "V"}) + _, err := p.SynthesizeStream(context.Background(), "hi", audio.TTSOptions{Model: "bogus-model"}) + if err == nil { + t.Fatal("expected error for unknown model, got nil") + } +} diff --git a/internal/audio/elevenlabs/tts_test.go b/internal/audio/elevenlabs/tts_test.go new file mode 100644 index 00000000..1ad44441 --- /dev/null +++ b/internal/audio/elevenlabs/tts_test.go @@ -0,0 +1,43 @@ +package elevenlabs + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/audio" +) + +func TestSynthesize_RejectsUnknownModel(t *testing.T) { + t.Parallel() + // Server must never be reached — validation runs before HTTP. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("unexpected upstream request for invalid model") + })) + defer srv.Close() + + p := NewTTSProvider(Config{APIKey: "k", BaseURL: srv.URL, VoiceID: "V"}) + _, err := p.Synthesize(context.Background(), "hi", audio.TTSOptions{Model: "bogus-model"}) + if err == nil { + t.Fatal("expected error for unknown model, got nil") + } +} + +func TestSynthesize_EmptyModelPassesValidation(t *testing.T) { + t.Parallel() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "audio/mpeg") + _, _ = w.Write([]byte("AUDIO")) + })) + defer srv.Close() + + p := NewTTSProvider(Config{APIKey: "k", BaseURL: srv.URL, VoiceID: "V"}) + res, err := p.Synthesize(context.Background(), "hi", audio.TTSOptions{}) + if err != nil { + t.Fatalf("Synthesize with empty model: %v", err) + } + if string(res.Audio) != "AUDIO" { + t.Errorf("audio: got %q, want AUDIO", res.Audio) + } +}