add openai realtime streaming adapter with reconnection logic
This commit is contained in:
leonardotrapani
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user