From e9499b0d6b38d81dd4b314db94e7142bfef236de Mon Sep 17 00:00:00 2001 From: leonardotrapani Date: Sun, 1 Feb 2026 01:47:12 +0100 Subject: [PATCH] add deepgram streaming adapter with reconnection logic --- internal/transcriber/adapter_deepgram.go | 406 ++++++++++++++++ internal/transcriber/adapter_deepgram_test.go | 434 ++++++++++++++++++ progress.txt | 15 +- tasks/prd.jsonc | 2 +- 4 files changed, 855 insertions(+), 2 deletions(-) create mode 100644 internal/transcriber/adapter_deepgram.go create mode 100644 internal/transcriber/adapter_deepgram_test.go diff --git a/internal/transcriber/adapter_deepgram.go b/internal/transcriber/adapter_deepgram.go new file mode 100644 index 0000000..fda830d --- /dev/null +++ b/internal/transcriber/adapter_deepgram.go @@ -0,0 +1,406 @@ +package transcriber + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "net/url" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/leonardotrapani/hyprvoice/internal/language" + "github.com/leonardotrapani/hyprvoice/internal/provider" +) + +// DeepgramAdapter implements StreamingAdapter for Deepgram real-time transcription +type DeepgramAdapter struct { + endpoint *provider.EndpointConfig + apiKey string + model string + language string + conn *websocket.Conn + resultsCh chan TranscriptionResult + mu sync.Mutex + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + started bool + + // reconnection config + maxRetries int + retryDelays []time.Duration +} + +// Deepgram WebSocket response types (incoming) +type deepgramWSResponse struct { + Type string `json:"type"` + Channel *deepgramChannel `json:"channel,omitempty"` + Metadata *deepgramMetadata `json:"metadata,omitempty"` + Error *deepgramError `json:"error,omitempty"` + ChannelIdx []int `json:"channel_index,omitempty"` + Duration float64 `json:"duration,omitempty"` + Start float64 `json:"start,omitempty"` + IsFinal bool `json:"is_final,omitempty"` + SpeechFinal bool `json:"speech_final,omitempty"` +} + +type deepgramChannel struct { + Alternatives []deepgramAlternative `json:"alternatives,omitempty"` +} + +type deepgramAlternative struct { + Transcript string `json:"transcript"` + Confidence float64 `json:"confidence"` +} + +type deepgramMetadata struct { + RequestID string `json:"request_id"` + ModelInfo struct { + Name string `json:"name"` + Version string `json:"version"` + } `json:"model_info"` +} + +type deepgramError struct { + Type string `json:"type"` + Message string `json:"message"` + Description string `json:"description,omitempty"` +} + +// NewDeepgramAdapter creates a new streaming adapter for Deepgram +// endpoint: the WebSocket endpoint config (e.g., wss://api.deepgram.com, /v1/listen) +// apiKey: Deepgram API key +// model: model ID (e.g., "nova-3") +// lang: canonical language code (will be converted to provider format) +func NewDeepgramAdapter(endpoint *provider.EndpointConfig, apiKey, model, lang string) *DeepgramAdapter { + return &DeepgramAdapter{ + endpoint: endpoint, + apiKey: apiKey, + model: model, + language: lang, + resultsCh: make(chan TranscriptionResult, 100), + maxRetries: 3, + retryDelays: defaultRetryDelays, + } +} + +// Start initiates the WebSocket connection to Deepgram +func (a *DeepgramAdapter) Start(ctx context.Context, lang string) error { + a.mu.Lock() + defer a.mu.Unlock() + + if a.started { + return fmt.Errorf("adapter already started") + } + + // use lang param if provided, otherwise use constructor lang + if lang != "" { + a.language = lang + } + + // create cancelable context + a.ctx, a.cancel = context.WithCancel(ctx) + + // connect to WebSocket + if err := a.connectLocked(); err != nil { + return err + } + a.started = true + + // start reader goroutine + a.wg.Add(1) + go a.readLoop() + + log.Printf("deepgram: connected, model=%s, language=%s", a.model, a.language) + return nil +} + +// connectLocked establishes WebSocket connection. Must be called with mu held. +func (a *DeepgramAdapter) connectLocked() error { + wsURL, err := a.buildURL() + if err != nil { + return fmt.Errorf("build websocket url: %w", err) + } + + headers := http.Header{} + headers.Set("Authorization", "Token "+a.apiKey) + + log.Printf("deepgram: connecting to %s", wsURL) + conn, resp, err := websocket.DefaultDialer.DialContext(a.ctx, wsURL, headers) + if err != nil { + if resp != nil { + log.Printf("deepgram: dial failed with status %d", resp.StatusCode) + } + return fmt.Errorf("websocket dial: %w", err) + } + a.conn = conn + return nil +} + +// reconnect attempts to re-establish the WebSocket connection with exponential backoff. +// Returns true if reconnection succeeded. +func (a *DeepgramAdapter) reconnect() bool { + for attempt := 0; attempt < a.maxRetries; attempt++ { + // check if context cancelled + select { + case <-a.ctx.Done(): + return false + default: + } + + // wait before retry (skip wait on first attempt) + if attempt > 0 { + delay := a.retryDelays[attempt-1] + if attempt-1 >= len(a.retryDelays) { + delay = a.retryDelays[len(a.retryDelays)-1] + } + log.Printf("deepgram: reconnect attempt %d/%d after %v", attempt+1, a.maxRetries, delay) + + select { + case <-a.ctx.Done(): + return false + case <-time.After(delay): + } + } else { + log.Printf("deepgram: reconnect attempt %d/%d", attempt+1, a.maxRetries) + } + + a.mu.Lock() + // close old connection if exists + if a.conn != nil { + a.conn.Close() + a.conn = nil + } + + err := a.connectLocked() + a.mu.Unlock() + + if err == nil { + log.Printf("deepgram: reconnected successfully") + // notify caller of brief interruption + select { + case a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("connection interrupted, reconnected"), IsFinal: false}: + default: + } + return true + } + + log.Printf("deepgram: reconnect failed: %v", err) + } + + return false +} + +// buildURL constructs the WebSocket URL with query parameters +func (a *DeepgramAdapter) buildURL() (string, error) { + // parse base URL and path + baseURL := a.endpoint.BaseURL + a.endpoint.Path + + u, err := url.Parse(baseURL) + if err != nil { + return "", fmt.Errorf("parse base url: %w", err) + } + + // add query parameters + q := u.Query() + q.Set("model", a.model) + q.Set("encoding", "linear16") // 16-bit linear PCM + q.Set("sample_rate", "16000") // 16kHz + q.Set("channels", "1") // mono + + // enable interim results + q.Set("interim_results", "true") + // enable smart formatting for better output + q.Set("smart_format", "true") + // enable punctuation + q.Set("punctuate", "true") + + // add language if specified + providerLang := language.ToProviderFormat(a.language, "deepgram") + if providerLang != "" { + q.Set("language", providerLang) + } + + u.RawQuery = q.Encode() + return u.String(), nil +} + +// readLoop reads messages from the WebSocket and sends results to the channel +func (a *DeepgramAdapter) readLoop() { + defer a.wg.Done() + defer close(a.resultsCh) + + for { + select { + case <-a.ctx.Done(): + return + default: + } + + a.mu.Lock() + conn := a.conn + a.mu.Unlock() + + if conn == nil { + // no connection, try to reconnect + if !a.reconnect() { + a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("connection lost, reconnection failed after %d attempts", a.maxRetries)} + return + } + continue + } + + _, message, err := conn.ReadMessage() + if err != nil { + // check if context was cancelled (normal shutdown) + select { + case <-a.ctx.Done(): + return + default: + } + + // attempt reconnection + log.Printf("deepgram: read error: %v, attempting reconnection", err) + if !a.reconnect() { + a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("websocket read: %w, reconnection failed", err)} + return + } + continue + } + + // parse message + var resp deepgramWSResponse + if err := json.Unmarshal(message, &resp); err != nil { + log.Printf("deepgram: parse error: %v", err) + continue + } + + // handle different message types + switch resp.Type { + case "Metadata": + if resp.Metadata != nil { + log.Printf("deepgram: session started, request_id=%s, model=%s", + resp.Metadata.RequestID, resp.Metadata.ModelInfo.Name) + } + + case "Results": + // transcription result + if resp.Channel != nil && len(resp.Channel.Alternatives) > 0 { + transcript := resp.Channel.Alternatives[0].Transcript + if transcript != "" { + isFinal := resp.IsFinal || resp.SpeechFinal + if isFinal { + log.Printf("deepgram: final: %q", transcript) + } + a.resultsCh <- TranscriptionResult{Text: transcript, IsFinal: isFinal} + } + } + + case "Error": + if resp.Error != nil { + errMsg := resp.Error.Message + if resp.Error.Description != "" { + errMsg = fmt.Sprintf("%s: %s", errMsg, resp.Error.Description) + } + log.Printf("deepgram: error: %s", errMsg) + a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("deepgram: %s", errMsg)} + } + + case "UtteranceEnd": + log.Printf("deepgram: utterance end detected") + + case "SpeechStarted": + log.Printf("deepgram: speech started") + + default: + log.Printf("deepgram: unknown message type: %s", resp.Type) + } + } +} + +// SendChunk sends audio data to the WebSocket +// Deepgram expects raw binary audio data, not base64 encoded +func (a *DeepgramAdapter) SendChunk(audio []byte) error { + a.mu.Lock() + if !a.started { + a.mu.Unlock() + return fmt.Errorf("adapter not started") + } + conn := a.conn + a.mu.Unlock() + + // check context + select { + case <-a.ctx.Done(): + return a.ctx.Err() + default: + } + + if conn == nil { + return fmt.Errorf("no connection") + } + + // send raw binary audio (not base64) + a.mu.Lock() + err := a.conn.WriteMessage(websocket.BinaryMessage, audio) + a.mu.Unlock() + + if err != nil { + // attempt reconnection + log.Printf("deepgram: write error: %v, attempting reconnection", err) + if a.reconnect() { + // retry the chunk after reconnection + a.mu.Lock() + err = a.conn.WriteMessage(websocket.BinaryMessage, audio) + a.mu.Unlock() + if err == nil { + return nil + } + } + return fmt.Errorf("websocket write: %w", err) + } + + return nil +} + +// Results returns the channel for receiving transcription results +func (a *DeepgramAdapter) Results() <-chan TranscriptionResult { + return a.resultsCh +} + +// Close gracefully closes the WebSocket connection +func (a *DeepgramAdapter) Close() error { + a.mu.Lock() + + if !a.started { + a.mu.Unlock() + return nil + } + + // cancel context first to signal reader to stop + if a.cancel != nil { + a.cancel() + } + + // get conn ref while holding lock + conn := a.conn + + a.started = false + a.mu.Unlock() + + // close websocket outside of lock (readLoop may be blocked on read) + if conn != nil { + // send close frame (best effort) + _ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) + conn.Close() + } + + // wait for reader to finish + a.wg.Wait() + + log.Printf("deepgram: closed") + return nil +} diff --git a/internal/transcriber/adapter_deepgram_test.go b/internal/transcriber/adapter_deepgram_test.go new file mode 100644 index 0000000..b31110f --- /dev/null +++ b/internal/transcriber/adapter_deepgram_test.go @@ -0,0 +1,434 @@ +package transcriber + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/leonardotrapani/hyprvoice/internal/provider" +) + +func TestDeepgramAdapter_ImplementsStreamingAdapter(t *testing.T) { + var _ StreamingAdapter = (*DeepgramAdapter)(nil) +} + +func TestDeepgramAdapter_Creation(t *testing.T) { + endpoint := &provider.EndpointConfig{ + BaseURL: "wss://api.deepgram.com", + Path: "/v1/listen", + } + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + if adapter.apiKey != "test-api-key" { + t.Errorf("apiKey = %q, want %q", adapter.apiKey, "test-api-key") + } + if adapter.model != "nova-3" { + t.Errorf("model = %q, want %q", adapter.model, "nova-3") + } + if adapter.language != "en" { + t.Errorf("language = %q, want %q", adapter.language, "en") + } + if adapter.maxRetries != 3 { + t.Errorf("maxRetries = %d, want %d", adapter.maxRetries, 3) + } +} + +func TestDeepgramAdapter_BuildURL(t *testing.T) { + tests := []struct { + name string + model string + language string + wantURL []string // URL must contain all these substrings + }{ + { + name: "english", + model: "nova-3", + language: "en", + wantURL: []string{"model=nova-3", "language=en-US", "encoding=linear16", "sample_rate=16000"}, + }, + { + name: "spanish", + model: "nova-2", + language: "es", + wantURL: []string{"model=nova-2", "language=es", "encoding=linear16"}, + }, + { + name: "auto-detect", + model: "nova-3", + language: "", + wantURL: []string{"model=nova-3", "encoding=linear16"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + endpoint := &provider.EndpointConfig{ + BaseURL: "wss://api.deepgram.com", + Path: "/v1/listen", + } + adapter := NewDeepgramAdapter(endpoint, "test-key", tt.model, tt.language) + + url, err := adapter.buildURL() + if err != nil { + t.Fatalf("buildURL() error = %v", err) + } + + for _, want := range tt.wantURL { + if !strings.Contains(url, want) { + t.Errorf("buildURL() = %q, want to contain %q", url, want) + } + } + }) + } +} + +func TestDeepgramAdapter_SendChunkNotStarted(t *testing.T) { + endpoint := &provider.EndpointConfig{ + BaseURL: "wss://api.deepgram.com", + Path: "/v1/listen", + } + adapter := NewDeepgramAdapter(endpoint, "test-key", "nova-3", "en") + + err := adapter.SendChunk([]byte("audio data")) + if err == nil { + t.Error("SendChunk() should return error when adapter not started") + } + if !strings.Contains(err.Error(), "not started") { + t.Errorf("error should mention 'not started', got: %v", err) + } +} + +func TestDeepgramAdapter_CloseNotStarted(t *testing.T) { + endpoint := &provider.EndpointConfig{ + BaseURL: "wss://api.deepgram.com", + Path: "/v1/listen", + } + adapter := NewDeepgramAdapter(endpoint, "test-key", "nova-3", "en") + + // closing not-started adapter should not error + err := adapter.Close() + if err != nil { + t.Errorf("Close() error = %v, want nil", err) + } +} + +// mockDeepgramServer creates a mock WebSocket server for testing +func mockDeepgramServer(t *testing.T, handler func(*websocket.Conn)) *httptest.Server { + upgrader := websocket.Upgrader{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // verify auth header + auth := r.Header.Get("Authorization") + if !strings.HasPrefix(auth, "Token ") { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Logf("upgrade error: %v", err) + return + } + defer conn.Close() + + handler(conn) + })) + return server +} + +func TestDeepgramAdapter_StartAndClose(t *testing.T) { + server := mockDeepgramServer(t, func(conn *websocket.Conn) { + // send metadata response + metadata := deepgramWSResponse{ + Type: "Metadata", + Metadata: &deepgramMetadata{ + RequestID: "test-123", + }, + } + metadata.Metadata.ModelInfo.Name = "nova-3" + if err := conn.WriteJSON(metadata); err != nil { + t.Logf("write metadata error: %v", err) + return + } + + // wait for close + for { + _, _, err := conn.ReadMessage() + if err != nil { + break + } + } + }) + defer server.Close() + + // convert http URL to ws URL + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{ + BaseURL: wsURL, + Path: "", + } + + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + ctx := context.Background() + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start() error = %v", err) + } + + // verify can't start twice + if err := adapter.Start(ctx, ""); err == nil { + t.Error("Start() should return error when already started") + } + + // close + if err := adapter.Close(); err != nil { + t.Errorf("Close() error = %v", err) + } +} + +func TestDeepgramAdapter_ReceivesResults(t *testing.T) { + var wg sync.WaitGroup + wg.Add(1) + + server := mockDeepgramServer(t, func(conn *websocket.Conn) { + defer wg.Done() + + // send metadata + metadata := deepgramWSResponse{Type: "Metadata", Metadata: &deepgramMetadata{RequestID: "test-123"}} + _ = conn.WriteJSON(metadata) + + // send interim result + interim := deepgramWSResponse{ + Type: "Results", + IsFinal: false, + Channel: &deepgramChannel{ + Alternatives: []deepgramAlternative{{Transcript: "hello", Confidence: 0.95}}, + }, + } + _ = conn.WriteJSON(interim) + + // send final result + final := deepgramWSResponse{ + Type: "Results", + IsFinal: true, + Channel: &deepgramChannel{ + Alternatives: []deepgramAlternative{{Transcript: "hello world", Confidence: 0.98}}, + }, + } + _ = conn.WriteJSON(final) + + // wait briefly then close + time.Sleep(50 * time.Millisecond) + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + ctx := context.Background() + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start() error = %v", err) + } + + // collect results + var results []TranscriptionResult + timeout := time.After(2 * time.Second) + +loop: + for { + select { + case result, ok := <-adapter.Results(): + if !ok { + break loop + } + results = append(results, result) + if result.IsFinal { + break loop + } + case <-timeout: + t.Fatal("timeout waiting for results") + } + } + + adapter.Close() + wg.Wait() + + // verify results + if len(results) < 2 { + t.Fatalf("expected at least 2 results, got %d", len(results)) + } + + // check interim + if results[0].Text != "hello" || results[0].IsFinal { + t.Errorf("interim result = %+v, want Text='hello', IsFinal=false", results[0]) + } + + // check final + found := false + for _, r := range results { + if r.Text == "hello world" && r.IsFinal { + found = true + break + } + } + if !found { + t.Errorf("did not find expected final result 'hello world'") + } +} + +func TestDeepgramAdapter_SendsRawBinaryAudio(t *testing.T) { + receivedAudio := make(chan []byte, 1) + + server := mockDeepgramServer(t, func(conn *websocket.Conn) { + // send metadata first + metadata := deepgramWSResponse{Type: "Metadata", Metadata: &deepgramMetadata{RequestID: "test-123"}} + _ = conn.WriteJSON(metadata) + + // read audio chunk + msgType, data, err := conn.ReadMessage() + if err != nil { + return + } + if msgType != websocket.BinaryMessage { + t.Errorf("expected binary message, got %d", msgType) + } + receivedAudio <- data + + // keep reading until close + for { + _, _, err := conn.ReadMessage() + if err != nil { + break + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + ctx := context.Background() + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start() error = %v", err) + } + + // send audio chunk + testAudio := []byte{0x01, 0x02, 0x03, 0x04} + if err := adapter.SendChunk(testAudio); err != nil { + t.Errorf("SendChunk() error = %v", err) + } + + // verify audio was received + select { + case audio := <-receivedAudio: + if string(audio) != string(testAudio) { + t.Errorf("received audio = %v, want %v", audio, testAudio) + } + case <-time.After(time.Second): + t.Error("timeout waiting for audio") + } + + adapter.Close() +} + +func TestDeepgramAdapter_HandlesError(t *testing.T) { + server := mockDeepgramServer(t, func(conn *websocket.Conn) { + // send error + errResp := deepgramWSResponse{ + Type: "Error", + Error: &deepgramError{ + Type: "AuthError", + Message: "Invalid API key", + }, + } + data, _ := json.Marshal(errResp) + _ = conn.WriteMessage(websocket.TextMessage, data) + + time.Sleep(50 * time.Millisecond) + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + ctx := context.Background() + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start() error = %v", err) + } + + // wait for error result + select { + case result := <-adapter.Results(): + if result.Error == nil { + t.Error("expected error result") + } + if !strings.Contains(result.Error.Error(), "Invalid API key") { + t.Errorf("error = %v, want to contain 'Invalid API key'", result.Error) + } + case <-time.After(time.Second): + t.Error("timeout waiting for error") + } + + adapter.Close() +} + +func TestDeepgramAdapter_ContextCancellation(t *testing.T) { + server := mockDeepgramServer(t, func(conn *websocket.Conn) { + // send metadata first so connection is established + metadata := deepgramWSResponse{Type: "Metadata", Metadata: &deepgramMetadata{RequestID: "test-123"}} + _ = conn.WriteJSON(metadata) + + // just keep connection open until closed + for { + _, _, err := conn.ReadMessage() + if err != nil { + break + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewDeepgramAdapter(endpoint, "test-api-key", "nova-3", "en") + + ctx, cancel := context.WithCancel(context.Background()) + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start() error = %v", err) + } + + // cancel context - this should trigger Close() to be called or at least stop SendChunk + cancel() + + // SendChunk should return error after context cancelled + err := adapter.SendChunk([]byte("test")) + if err == nil { + // it's ok if first chunk after cancel succeeds - the context cancel is async + // but subsequent operations should fail + } + + // Close should work even after context cancelled + if err := adapter.Close(); err != nil { + t.Errorf("Close() error = %v", err) + } + + // results channel should be closed after Close() + select { + case _, ok := <-adapter.Results(): + if ok { + // drain any remaining + for range adapter.Results() { + } + } + case <-time.After(2 * time.Second): + t.Error("timeout waiting for results channel to close") + } +} diff --git a/progress.txt b/progress.txt index 06efa9c..6e9fe69 100644 --- a/progress.txt +++ b/progress.txt @@ -346,4 +346,17 @@ Started: Sun Feb 1 12:22:47 AM CET 2026 - DefaultModel(Transcription) returns 'nova-3' - Registered in provider.init() - Comprehensive test file created: deepgram_test.go -- All tests passing, typecheck passes \ No newline at end of file +- All tests passing, typecheck passes + +### Task 34: Create Deepgram StreamingAdapter +- Created `internal/transcriber/adapter_deepgram.go` +- DeepgramAdapter struct: endpoint, apiKey, model, language, conn, resultsCh, mutex, ctx/cancel, WaitGroup +- Start(): connects to wss://api.deepgram.com/v1/listen with Authorization: Token header +- Query params: model, language, encoding=linear16, sample_rate=16000, channels=1, interim_results=true, smart_format=true, punctuate=true +- Language conversion via language.ToProviderFormat(lang, "deepgram") +- SendChunk(): sends raw binary audio (websocket.BinaryMessage, not base64 like ElevenLabs) +- readLoop goroutine: parses Metadata, Results (interim + final), Error, UtteranceEnd, SpeechStarted messages +- Close(): cancels context, sends close frame, waits for reader goroutine +- Added reconnection logic (maxRetries=3, exponential backoff 1s, 2s, 4s) matching ElevenLabs pattern +- Comprehensive tests with mock WebSocket server +- All tests passing with -race flag, typecheck passes \ No newline at end of file diff --git a/tasks/prd.jsonc b/tasks/prd.jsonc index c3caa61..f54e276 100644 --- a/tasks/prd.jsonc +++ b/tasks/prd.jsonc @@ -806,7 +806,7 @@ "Close() terminates cleanly", "Typecheck passes" ], - "passes": false + "passes": true }, { "title": "Add reconnection logic to Deepgram StreamingAdapter",