diff --git a/internal/transcriber/adapter_openai_realtime.go b/internal/transcriber/adapter_openai_realtime.go new file mode 100644 index 0000000..8de1b65 --- /dev/null +++ b/internal/transcriber/adapter_openai_realtime.go @@ -0,0 +1,535 @@ +package transcriber + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "log" + "net/http" + "net/url" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/leonardotrapani/hyprvoice/internal/provider" +) + +// OpenAIRealtimeAdapter implements StreamingAdapter for OpenAI Realtime API transcription +type OpenAIRealtimeAdapter 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 + + // track current item for transcription + currentItemID string +} + +// OpenAI Realtime WebSocket message types (outgoing) +type openaiRealtimeSessionUpdate struct { + Type string `json:"type"` + Session openaiRealtimeSessionConfig `json:"session"` +} + +type openaiRealtimeSessionConfig struct { + Modalities []string `json:"modalities,omitempty"` + InputAudioFormat string `json:"input_audio_format,omitempty"` + InputAudioTranscription *openaiRealtimeTranscription `json:"input_audio_transcription,omitempty"` + TurnDetection *openaiRealtimeTurnDetection `json:"turn_detection,omitempty"` +} + +type openaiRealtimeTranscription struct { + Model string `json:"model,omitempty"` + Language string `json:"language,omitempty"` +} + +type openaiRealtimeTurnDetection struct { + Type string `json:"type"` + Threshold float64 `json:"threshold,omitempty"` + PrefixPaddingMs int `json:"prefix_padding_ms,omitempty"` + SilenceDurationMs int `json:"silence_duration_ms,omitempty"` + CreateResponse bool `json:"create_response,omitempty"` +} + +type openaiRealtimeInputAudioAppend struct { + Type string `json:"type"` + Audio string `json:"audio"` +} + +type openaiRealtimeInputAudioCommit struct { + Type string `json:"type"` +} + +// OpenAI Realtime WebSocket response types (incoming) +type openaiRealtimeServerEvent struct { + Type string `json:"type"` + EventID string `json:"event_id,omitempty"` + Session *openaiRealtimeSessionInfo `json:"session,omitempty"` + Error *openaiRealtimeError `json:"error,omitempty"` + ItemID string `json:"item_id,omitempty"` + ContentIndex int `json:"content_index,omitempty"` + Transcript string `json:"transcript,omitempty"` + Delta string `json:"delta,omitempty"` +} + +type openaiRealtimeSessionInfo struct { + ID string `json:"id"` + Model string `json:"model"` +} + +type openaiRealtimeError struct { + Type string `json:"type"` + Code string `json:"code,omitempty"` + Message string `json:"message"` + Param string `json:"param,omitempty"` +} + +// NewOpenAIRealtimeAdapter creates a new streaming adapter for OpenAI Realtime API +// endpoint: the WebSocket endpoint config (e.g., wss://api.openai.com, /v1/realtime) +// apiKey: OpenAI API key +// model: model ID (e.g., "gpt-4o-realtime-preview") +// lang: canonical language code (will be used for transcription config) +func NewOpenAIRealtimeAdapter(endpoint *provider.EndpointConfig, apiKey, model, lang string) *OpenAIRealtimeAdapter { + return &OpenAIRealtimeAdapter{ + endpoint: endpoint, + apiKey: apiKey, + model: model, + language: lang, + resultsCh: make(chan TranscriptionResult, 100), + maxRetries: 3, + retryDelays: defaultRetryDelays, + } +} + +// Start initiates the WebSocket connection to OpenAI Realtime API +func (a *OpenAIRealtimeAdapter) 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("openai-realtime: connected, model=%s, language=%s", a.model, a.language) + return nil +} + +// connectLocked establishes WebSocket connection and configures session. Must be called with mu held. +func (a *OpenAIRealtimeAdapter) connectLocked() error { + wsURL, err := a.buildURL() + if err != nil { + return fmt.Errorf("build websocket url: %w", err) + } + + headers := http.Header{} + headers.Set("Authorization", "Bearer "+a.apiKey) + headers.Set("OpenAI-Beta", "realtime=v1") + + log.Printf("openai-realtime: connecting to %s", wsURL) + conn, resp, err := websocket.DefaultDialer.DialContext(a.ctx, wsURL, headers) + if err != nil { + if resp != nil { + log.Printf("openai-realtime: dial failed with status %d", resp.StatusCode) + } + return fmt.Errorf("websocket dial: %w", err) + } + a.conn = conn + + // configure session for transcription-only mode + if err := a.configureSession(); err != nil { + conn.Close() + a.conn = nil + return fmt.Errorf("configure session: %w", err) + } + + return nil +} + +// configureSession sends session.update to configure transcription mode +func (a *OpenAIRealtimeAdapter) configureSession() error { + // configure for transcription-only mode + // use server VAD to automatically detect speech and commit audio + sessionUpdate := openaiRealtimeSessionUpdate{ + Type: "session.update", + Session: openaiRealtimeSessionConfig{ + Modalities: []string{"text"}, // text only, no audio output + InputAudioFormat: "pcm16", // we send 16-bit PCM + InputAudioTranscription: &openaiRealtimeTranscription{ + Model: "gpt-4o-transcribe", // use gpt-4o for input transcription + }, + TurnDetection: &openaiRealtimeTurnDetection{ + Type: "server_vad", + Threshold: 0.5, + PrefixPaddingMs: 300, + SilenceDurationMs: 500, + CreateResponse: false, // we don't want responses, just transcription + }, + }, + } + + // add language if specified + if a.language != "" { + sessionUpdate.Session.InputAudioTranscription.Language = a.language + } + + return a.conn.WriteJSON(sessionUpdate) +} + +// reconnect attempts to re-establish the WebSocket connection with exponential backoff. +// Returns true if reconnection succeeded. +func (a *OpenAIRealtimeAdapter) 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("openai-realtime: 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("openai-realtime: 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("openai-realtime: 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("openai-realtime: reconnect failed: %v", err) + } + + return false +} + +// buildURL constructs the WebSocket URL with query parameters +func (a *OpenAIRealtimeAdapter) 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 model as query parameter + q := u.Query() + q.Set("model", a.model) + u.RawQuery = q.Encode() + + return u.String(), nil +} + +// readLoop reads messages from the WebSocket and sends results to the channel +func (a *OpenAIRealtimeAdapter) 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("openai-realtime: 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 event openaiRealtimeServerEvent + if err := json.Unmarshal(message, &event); err != nil { + log.Printf("openai-realtime: parse error: %v", err) + continue + } + + // handle different event types + a.handleEvent(event) + } +} + +// handleEvent processes incoming server events +func (a *OpenAIRealtimeAdapter) handleEvent(event openaiRealtimeServerEvent) { + switch event.Type { + case "session.created": + if event.Session != nil { + log.Printf("openai-realtime: session created, id=%s, model=%s", event.Session.ID, event.Session.Model) + } + + case "session.updated": + log.Printf("openai-realtime: session updated") + + case "error": + if event.Error != nil { + errMsg := event.Error.Message + if event.Error.Code != "" { + errMsg = fmt.Sprintf("%s: %s", event.Error.Code, errMsg) + } + log.Printf("openai-realtime: error: %s", errMsg) + a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("openai: %s", errMsg)} + } + + case "input_audio_buffer.speech_started": + log.Printf("openai-realtime: speech started") + + case "input_audio_buffer.speech_stopped": + log.Printf("openai-realtime: speech stopped, item_id=%s", event.ItemID) + a.currentItemID = event.ItemID + + case "input_audio_buffer.committed": + log.Printf("openai-realtime: audio committed, item_id=%s", event.ItemID) + a.currentItemID = event.ItemID + + case "conversation.item.input_audio_transcription.delta": + // partial transcription result + if event.Delta != "" { + a.resultsCh <- TranscriptionResult{Text: event.Delta, IsFinal: false} + } + + case "conversation.item.input_audio_transcription.completed": + // final transcription result + if event.Transcript != "" { + log.Printf("openai-realtime: transcription completed: %q", event.Transcript) + a.resultsCh <- TranscriptionResult{Text: event.Transcript, IsFinal: true} + } + + case "conversation.item.input_audio_transcription.failed": + log.Printf("openai-realtime: transcription failed for item %s", event.ItemID) + if event.Error != nil { + a.resultsCh <- TranscriptionResult{Error: fmt.Errorf("transcription failed: %s", event.Error.Message)} + } + + case "conversation.item.created", "conversation.item.added": + log.Printf("openai-realtime: conversation item created/added") + + case "rate_limits.updated": + // ignore rate limit updates + + default: + log.Printf("openai-realtime: unhandled event type: %s", event.Type) + } +} + +// SendChunk sends audio data to the WebSocket +// OpenAI Realtime API expects base64-encoded PCM16 audio at 24kHz +// We receive 16kHz audio, so we need to resample +func (a *OpenAIRealtimeAdapter) 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") + } + + // resample from 16kHz to 24kHz (OpenAI expects 24kHz) + resampled := resample16to24(audio) + + // encode audio as base64 + audioB64 := base64.StdEncoding.EncodeToString(resampled) + + // create message + msg := openaiRealtimeInputAudioAppend{ + Type: "input_audio_buffer.append", + Audio: audioB64, + } + + // send as JSON + a.mu.Lock() + err := a.conn.WriteJSON(msg) + a.mu.Unlock() + + if err != nil { + // attempt reconnection + log.Printf("openai-realtime: write error: %v, attempting reconnection", err) + if a.reconnect() { + // retry the chunk after reconnection + a.mu.Lock() + err = a.conn.WriteJSON(msg) + a.mu.Unlock() + if err == nil { + return nil + } + } + return fmt.Errorf("websocket write: %w", err) + } + + return nil +} + +// resample16to24 converts 16kHz PCM16 audio to 24kHz using linear interpolation +// Input: 16-bit PCM samples at 16kHz +// Output: 16-bit PCM samples at 24kHz +func resample16to24(input []byte) []byte { + if len(input) < 2 { + return input + } + + // input has 16kHz samples (2 bytes each) + // output needs 24kHz samples (ratio 24/16 = 1.5) + numInputSamples := len(input) / 2 + numOutputSamples := (numInputSamples * 3) / 2 + + output := make([]byte, numOutputSamples*2) + + for i := 0; i < numOutputSamples; i++ { + // calculate position in input + srcPos := float64(i) * 16.0 / 24.0 + srcIdx := int(srcPos) + frac := srcPos - float64(srcIdx) + + // get source samples + var sample1, sample2 int16 + if srcIdx*2+1 < len(input) { + sample1 = int16(input[srcIdx*2]) | (int16(input[srcIdx*2+1]) << 8) + } + if (srcIdx+1)*2+1 < len(input) { + sample2 = int16(input[(srcIdx+1)*2]) | (int16(input[(srcIdx+1)*2+1]) << 8) + } else { + sample2 = sample1 + } + + // linear interpolation + outSample := int16(float64(sample1)*(1-frac) + float64(sample2)*frac) + + // write output sample (little-endian) + output[i*2] = byte(outSample) + output[i*2+1] = byte(outSample >> 8) + } + + return output +} + +// Results returns the channel for receiving transcription results +func (a *OpenAIRealtimeAdapter) Results() <-chan TranscriptionResult { + return a.resultsCh +} + +// Close gracefully closes the WebSocket connection +func (a *OpenAIRealtimeAdapter) 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("openai-realtime: closed") + return nil +} diff --git a/internal/transcriber/adapter_openai_realtime_test.go b/internal/transcriber/adapter_openai_realtime_test.go new file mode 100644 index 0000000..be0c5ca --- /dev/null +++ b/internal/transcriber/adapter_openai_realtime_test.go @@ -0,0 +1,544 @@ +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" +) + +// mockOpenAIRealtimeServer creates a mock WebSocket server for OpenAI Realtime API +func mockOpenAIRealtimeServer(t *testing.T, handler func(*websocket.Conn)) *httptest.Server { + upgrader := websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { return true }, + } + + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // verify auth header + auth := r.Header.Get("Authorization") + if !strings.HasPrefix(auth, "Bearer ") { + t.Errorf("expected Bearer auth header, got: %s", auth) + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + // verify model in query + model := r.URL.Query().Get("model") + if model == "" { + t.Error("expected model query parameter") + } + + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade failed: %v", err) + return + } + defer conn.Close() + + handler(conn) + })) +} + +func TestOpenAIRealtimeAdapter_ImplementsInterface(t *testing.T) { + var _ StreamingAdapter = (*OpenAIRealtimeAdapter)(nil) +} + +func TestOpenAIRealtimeAdapter_Start(t *testing.T) { + var mu sync.Mutex + sessionCreated := false + sessionUpdated := false + + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + // send session.created event + sessionCreatedEvent := map[string]interface{}{ + "type": "session.created", + "event_id": "event_123", + "session": map[string]interface{}{ + "id": "sess_123", + "model": "gpt-4o-realtime-preview", + }, + } + if err := conn.WriteJSON(sessionCreatedEvent); err != nil { + t.Errorf("write session.created: %v", err) + } + mu.Lock() + sessionCreated = true + mu.Unlock() + + // read session.update from client + _, msg, err := conn.ReadMessage() + if err != nil { + return + } + + var update map[string]interface{} + if err := json.Unmarshal(msg, &update); err != nil { + t.Errorf("unmarshal session.update: %v", err) + return + } + + if update["type"] != "session.update" { + t.Errorf("expected session.update, got %s", update["type"]) + } + mu.Lock() + sessionUpdated = true + mu.Unlock() + + // send session.updated response + sessionUpdatedEvent := map[string]interface{}{ + "type": "session.updated", + "event_id": "event_124", + } + if err := conn.WriteJSON(sessionUpdatedEvent); err != nil { + t.Errorf("write session.updated: %v", err) + } + + // keep connection open until client closes + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + // extract host for endpoint + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + endpoint := &provider.EndpointConfig{ + BaseURL: wsURL, + Path: "", + } + + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test-key", "gpt-4o-realtime-preview", "en") + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + err := adapter.Start(ctx, "") + if err != nil { + t.Fatalf("Start failed: %v", err) + } + defer adapter.Close() + + // give time for events to process + time.Sleep(100 * time.Millisecond) + + mu.Lock() + created := sessionCreated + updated := sessionUpdated + mu.Unlock() + + if !created { + t.Error("session.created was not sent") + } + if !updated { + t.Error("session.update was not received by server") + } +} + +func TestOpenAIRealtimeAdapter_SendChunk(t *testing.T) { + var mu sync.Mutex + receivedAudio := false + + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + // send session.created + sessionCreatedEvent := map[string]interface{}{ + "type": "session.created", + "session": map[string]interface{}{ + "id": "sess_123", + }, + } + conn.WriteJSON(sessionCreatedEvent) + + // read session.update + conn.ReadMessage() + + // send session.updated + conn.WriteJSON(map[string]interface{}{"type": "session.updated"}) + + // read audio chunk + _, msg, err := conn.ReadMessage() + if err != nil { + return + } + + var audioMsg map[string]interface{} + if err := json.Unmarshal(msg, &audioMsg); err != nil { + t.Errorf("unmarshal audio: %v", err) + return + } + + if audioMsg["type"] == "input_audio_buffer.append" { + audio, ok := audioMsg["audio"].(string) + if ok && len(audio) > 0 { + mu.Lock() + receivedAudio = true + mu.Unlock() + } + } + + // keep connection open + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test", "gpt-4o-realtime-preview", "") + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start failed: %v", err) + } + defer adapter.Close() + + // give time for connection setup + time.Sleep(100 * time.Millisecond) + + // send audio chunk (16kHz PCM16) + audio := make([]byte, 320) // 10ms of 16kHz audio + for i := range audio { + audio[i] = byte(i % 256) + } + + if err := adapter.SendChunk(audio); err != nil { + t.Fatalf("SendChunk failed: %v", err) + } + + // give time for message to be sent + time.Sleep(100 * time.Millisecond) + + mu.Lock() + received := receivedAudio + mu.Unlock() + + if !received { + t.Error("server did not receive audio chunk") + } +} + +func TestOpenAIRealtimeAdapter_TranscriptionResults(t *testing.T) { + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + // send session.created + conn.WriteJSON(map[string]interface{}{"type": "session.created", "session": map[string]interface{}{"id": "sess_123"}}) + + // read session.update + conn.ReadMessage() + + // send session.updated + conn.WriteJSON(map[string]interface{}{"type": "session.updated"}) + + // simulate transcription events + time.Sleep(50 * time.Millisecond) + + // speech started + conn.WriteJSON(map[string]interface{}{ + "type": "input_audio_buffer.speech_started", + }) + + // partial transcription + conn.WriteJSON(map[string]interface{}{ + "type": "conversation.item.input_audio_transcription.delta", + "delta": "Hello", + }) + + // more partial + conn.WriteJSON(map[string]interface{}{ + "type": "conversation.item.input_audio_transcription.delta", + "delta": " world", + }) + + // final transcription + conn.WriteJSON(map[string]interface{}{ + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "Hello world", + }) + + // keep connection open + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test", "gpt-4o-realtime-preview", "en") + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start failed: %v", err) + } + defer adapter.Close() + + results := adapter.Results() + + // collect results + var partials []string + var finals []string + timeout := time.After(2 * time.Second) + + for { + select { + case result, ok := <-results: + if !ok { + goto done + } + if result.Error != nil { + continue + } + if result.IsFinal { + finals = append(finals, result.Text) + } else { + partials = append(partials, result.Text) + } + if len(finals) > 0 { + goto done + } + case <-timeout: + goto done + } + } +done: + + if len(partials) != 2 { + t.Errorf("expected 2 partial results, got %d: %v", len(partials), partials) + } + + if len(finals) != 1 { + t.Errorf("expected 1 final result, got %d: %v", len(finals), finals) + } + + if len(finals) > 0 && finals[0] != "Hello world" { + t.Errorf("expected final 'Hello world', got %q", finals[0]) + } +} + +func TestOpenAIRealtimeAdapter_ErrorHandling(t *testing.T) { + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + // send session.created + conn.WriteJSON(map[string]interface{}{"type": "session.created", "session": map[string]interface{}{"id": "sess_123"}}) + + // read session.update + conn.ReadMessage() + + // send session.updated + conn.WriteJSON(map[string]interface{}{"type": "session.updated"}) + + // send error event + time.Sleep(50 * time.Millisecond) + conn.WriteJSON(map[string]interface{}{ + "type": "error", + "error": map[string]interface{}{ + "type": "invalid_request_error", + "code": "invalid_audio", + "message": "Audio format is invalid", + }, + }) + + // keep connection open + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test", "gpt-4o-realtime-preview", "") + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start failed: %v", err) + } + defer adapter.Close() + + results := adapter.Results() + + // wait for error + select { + case result := <-results: + if result.Error == nil { + t.Error("expected error result") + } + if !strings.Contains(result.Error.Error(), "invalid_audio") { + t.Errorf("expected error containing 'invalid_audio', got: %v", result.Error) + } + case <-time.After(2 * time.Second): + t.Error("timeout waiting for error result") + } +} + +func TestOpenAIRealtimeAdapter_Reconnection(t *testing.T) { + connectCount := 0 + var mu sync.Mutex + + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + mu.Lock() + connectCount++ + count := connectCount + mu.Unlock() + + // send session.created + conn.WriteJSON(map[string]interface{}{"type": "session.created", "session": map[string]interface{}{"id": "sess_" + string(rune('0'+count))}}) + + // read session.update + conn.ReadMessage() + + // send session.updated + conn.WriteJSON(map[string]interface{}{"type": "session.updated"}) + + // first connection: close immediately to trigger reconnect + if count == 1 { + time.Sleep(50 * time.Millisecond) + conn.Close() + return + } + + // second connection: stay open + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test", "gpt-4o-realtime-preview", "") + adapter.retryDelays = []time.Duration{10 * time.Millisecond, 20 * time.Millisecond, 40 * time.Millisecond} + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start failed: %v", err) + } + defer adapter.Close() + + // wait for reconnection + time.Sleep(500 * time.Millisecond) + + mu.Lock() + finalCount := connectCount + mu.Unlock() + + if finalCount < 2 { + t.Errorf("expected at least 2 connections (reconnection), got %d", finalCount) + } +} + +func TestOpenAIRealtimeAdapter_Close(t *testing.T) { + server := mockOpenAIRealtimeServer(t, func(conn *websocket.Conn) { + conn.WriteJSON(map[string]interface{}{"type": "session.created", "session": map[string]interface{}{"id": "sess_123"}}) + conn.ReadMessage() + conn.WriteJSON(map[string]interface{}{"type": "session.updated"}) + + for { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + }) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + endpoint := &provider.EndpointConfig{BaseURL: wsURL, Path: ""} + adapter := NewOpenAIRealtimeAdapter(endpoint, "sk-test", "gpt-4o-realtime-preview", "") + + ctx := context.Background() + + if err := adapter.Start(ctx, ""); err != nil { + t.Fatalf("Start failed: %v", err) + } + + // close should not block or panic + err := adapter.Close() + if err != nil { + t.Errorf("Close returned error: %v", err) + } + + // results channel should be closed + select { + case _, ok := <-adapter.Results(): + if ok { + // drain any remaining results + for range adapter.Results() { + } + } + case <-time.After(time.Second): + t.Error("results channel not closed after Close()") + } +} + +func TestResample16to24(t *testing.T) { + // test with simple audio data + input := make([]byte, 32) // 16 samples at 16kHz + for i := 0; i < 16; i++ { + // write sample value (little-endian) + sample := int16(i * 1000) + input[i*2] = byte(sample) + input[i*2+1] = byte(sample >> 8) + } + + output := resample16to24(input) + + // 16 samples at 16kHz = 24 samples at 24kHz (ratio 1.5) + expectedSamples := 24 + if len(output) != expectedSamples*2 { + t.Errorf("expected %d bytes, got %d", expectedSamples*2, len(output)) + } + + // output should have reasonable values (interpolated) + for i := 0; i < expectedSamples; i++ { + sample := int16(output[i*2]) | (int16(output[i*2+1]) << 8) + if sample < -32768 || sample > 32767 { + t.Errorf("sample %d out of range: %d", i, sample) + } + } +} + +func TestResample16to24_EmptyInput(t *testing.T) { + output := resample16to24([]byte{}) + if len(output) != 0 { + t.Errorf("expected empty output for empty input, got %d bytes", len(output)) + } +} + +func TestResample16to24_SingleSample(t *testing.T) { + input := []byte{0x00, 0x10} // single sample + output := resample16to24(input) + // with only 1 sample, output should be minimal + if len(output) == 0 { + t.Error("expected non-empty output for single sample") + } +} diff --git a/progress.txt b/progress.txt index ed39f91..242c613 100644 --- a/progress.txt +++ b/progress.txt @@ -378,4 +378,20 @@ Started: Sun Feb 1 12:22:47 AM CET 2026 - DefaultModel(Transcription) unchanged (still returns 'whisper-1') - Added TestOpenAIRealtimeModel test verifying all properties - Updated TestModelsOfType to expect 4 transcription models for OpenAI -- All tests passing, typecheck passes \ No newline at end of file +- All tests passing, typecheck passes + +### Task 37: Create OpenAI Realtime StreamingAdapter +- Created `internal/transcriber/adapter_openai_realtime.go` +- OpenAIRealtimeAdapter struct: endpoint, apiKey, model, language, conn, resultsCh, mu, ctx/cancel, WaitGroup +- Start(): connects to wss://api.openai.com/v1/realtime?model=X with Bearer auth and OpenAI-Beta header +- Sends session.update to configure transcription-only mode (modalities=['text'], input_audio_format='pcm16') +- Enables input_audio_transcription with gpt-4o-transcribe model +- Uses server_vad turn detection for automatic speech detection +- SendChunk(): resamples audio from 16kHz to 24kHz, sends input_audio_buffer.append with base64 audio +- readLoop goroutine: parses conversation.item.input_audio_transcription.delta (partial) and .completed (final) +- Handles error events, speech_started, speech_stopped, session events +- Close(): cancels context, sends close frame, waits for reader goroutine +- Added resample16to24() for 16kHz to 24kHz PCM conversion using linear interpolation +- Added reconnection logic (maxRetries=3, exponential backoff 1s, 2s, 4s) - same pattern as ElevenLabs/Deepgram +- 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 45aa5c4..fe3445b 100644 --- a/tasks/prd.jsonc +++ b/tasks/prd.jsonc @@ -869,7 +869,7 @@ "Close() terminates cleanly", "Typecheck passes" ], - "passes": false + "passes": true }, { "title": "Add reconnection logic to OpenAI Realtime StreamingAdapter",