diff --git a/docs/features/tui/index.md b/docs/features/tui/index.md index fe85eaac8f..adb982769f 100644 --- a/docs/features/tui/index.md +++ b/docs/features/tui/index.md @@ -127,7 +127,9 @@ export OPENAI_API_KEY=your_openai_api_key docker agent run agent.yaml ``` -Run `/speak` to start voice input. Press Enter to stop and send the transcribed message, or Escape to stop without sending it. +Run `/speak` to start voice input. Press Enter to finish transcription and send the message once the final text arrives. Press Escape to cancel without sending; text already received stays in the draft. Errors or a finalization timeout keep the draft instead of sending a partial message. + +Dictation stays with the tab where it started. Switching tabs while transcription finishes keeps the text in the original draft without sending it. ### Plans Sidebar diff --git a/pkg/audio/capture/capture.go b/pkg/audio/capture/capture.go new file mode 100644 index 0000000000..b2a0700902 --- /dev/null +++ b/pkg/audio/capture/capture.go @@ -0,0 +1,26 @@ +// Package capture provides microphone capture using CoreAudio on macOS. +package capture + +import "errors" + +// Common sample rates. +const ( + SampleRate44100 = 44100 // CD quality, used for WAV recording + SampleRate24000 = 24000 // Required by OpenAI Realtime API +) + +// ErrNotSupported is returned when audio capture is unavailable on the current platform. +var ErrNotSupported = errors.New("audio capture is only supported on macOS") + +// AudioHandler receives mono, 16-bit signed PCM (little-endian) audio chunks. +type AudioHandler func(data []byte) + +// Capturer configures independent microphone capture runs. +type Capturer struct { + sampleRate int +} + +// NewCapturer creates a capturer with the specified sample rate. +func NewCapturer(sampleRate int) *Capturer { + return &Capturer{sampleRate: sampleRate} +} diff --git a/pkg/audio/capture/capture_darwin.go b/pkg/audio/capture/capture_darwin.go index 65533edf8c..0d312dca56 100644 --- a/pkg/audio/capture/capture_darwin.go +++ b/pkg/audio/capture/capture_darwin.go @@ -1,14 +1,15 @@ -//go:build darwin && !no_audio +//go:build darwin && cgo && !no_audio -// Package capture provides audio capture functionality for macOS using CoreAudio. package capture /* -#cgo CFLAGS: -x objective-c #cgo LDFLAGS: -framework CoreAudio -framework AudioToolbox -framework CoreFoundation #include -#include +#include +#include +#include +#include #include #include @@ -16,274 +17,209 @@ package capture #define BITS_PER_SAMPLE 16 #define NUM_BUFFERS 3 -// Capture state -static AudioQueueRef audioQueue = NULL; -static AudioQueueBufferRef buffers[NUM_BUFFERS]; -static AudioFileID audioFile = NULL; -static SInt64 currentPacket = 0; -static volatile int isCapturing = 0; -static double sampleRate = 44100.0; -static UInt32 bufferSize = 4096; +typedef struct { + AudioQueueRef queue; + AudioFileID file; + AudioStreamBasicDescription format; + SInt64 packet; + uintptr_t handle; + pthread_mutex_t mutex; + bool started; + bool stopping; + OSStatus error; +} Capture; -// Audio format description -static AudioStreamBasicDescription audioFormat; +void goAudioCallback(uintptr_t handle, void *data, int32_t size, int32_t status); -// Forward declaration for Go callback -void goAudioCallback(void *data, int size); - -// Callback function for audio input static void AudioInputCallback( - void *inUserData, - AudioQueueRef inAQ, - AudioQueueBufferRef inBuffer, - const AudioTimeStamp *inStartTime, - UInt32 inNumberPacketDescriptions, - const AudioStreamPacketDescription *inPacketDescs + void *userData, + AudioQueueRef queue, + AudioQueueBufferRef buffer, + const AudioTimeStamp *startTime, + UInt32 packetCount, + const AudioStreamPacketDescription *packetDescriptions ) { - if (!isCapturing || inBuffer->mAudioDataByteSize == 0) return; + Capture *capture = userData; + pthread_mutex_lock(&capture->mutex); + if (capture->error != noErr) { + pthread_mutex_unlock(&capture->mutex); + return; + } - // Write to file if recording - if (audioFile != NULL) { - UInt32 numPackets = inBuffer->mAudioDataByteSize / audioFormat.mBytesPerPacket; - if (AudioFileWritePackets(audioFile, false, inBuffer->mAudioDataByteSize, - NULL, currentPacket, &numPackets, inBuffer->mAudioData) == noErr) { - currentPacket += numPackets; + if (buffer->mAudioDataByteSize > 0) { + if (capture->file != NULL) { + UInt32 packets = buffer->mAudioDataByteSize / capture->format.mBytesPerPacket; + capture->error = AudioFileWritePackets(capture->file, false, + buffer->mAudioDataByteSize, NULL, capture->packet, &packets, buffer->mAudioData); + capture->packet += packets; + } + if (capture->error == noErr) { + goAudioCallback(capture->handle, buffer->mAudioData, (int)buffer->mAudioDataByteSize, noErr); } } - // Call Go callback for streaming - goAudioCallback(inBuffer->mAudioData, (int)inBuffer->mAudioDataByteSize); - - if (isCapturing) { - AudioQueueEnqueueBuffer(inAQ, inBuffer, 0, NULL); + if (!capture->stopping && capture->error == noErr) { + capture->error = AudioQueueEnqueueBuffer(queue, buffer, 0, NULL); } + if (capture->error != noErr) { + goAudioCallback(capture->handle, NULL, 0, capture->error); + } + pthread_mutex_unlock(&capture->mutex); } -// Initialize audio format with given sample rate -static void initAudioFormat(double rate) { - sampleRate = rate; - bufferSize = (UInt32)(rate * 0.1) * (BITS_PER_SAMPLE / 8) * CHANNELS; // 100ms buffer - - memset(&audioFormat, 0, sizeof(audioFormat)); - audioFormat.mSampleRate = rate; - audioFormat.mFormatID = kAudioFormatLinearPCM; - audioFormat.mFormatFlags = kLinearPCMFormatFlagIsSignedInteger | kLinearPCMFormatFlagIsPacked; - audioFormat.mFramesPerPacket = 1; - audioFormat.mChannelsPerFrame = CHANNELS; - audioFormat.mBitsPerChannel = BITS_PER_SAMPLE; - audioFormat.mBytesPerPacket = (BITS_PER_SAMPLE / 8) * CHANNELS; - audioFormat.mBytesPerFrame = (BITS_PER_SAMPLE / 8) * CHANNELS; +static OSStatus freeCapture(Capture *capture) { + pthread_mutex_lock(&capture->mutex); + capture->stopping = true; + pthread_mutex_unlock(&capture->mutex); + + OSStatus status = noErr; + if (capture->queue != NULL) { + // Stop delivers partially filled buffers before disposal joins callbacks. + if (capture->started) status = AudioQueueStop(capture->queue, true); + OSStatus disposeStatus = AudioQueueDispose(capture->queue, true); + if (status == noErr) status = disposeStatus; + } + if (capture->file != NULL) { + OSStatus fileStatus = AudioFileClose(capture->file); + if (status == noErr) status = fileStatus; + } + if (status == noErr) status = capture->error; + pthread_mutex_destroy(&capture->mutex); + free(capture); + return status; } -// Start capturing audio, optionally recording to file -// filePath can be NULL for streaming-only mode -static int startCapture(const char *filePath, double rate) { - if (isCapturing) return -1; - - initAudioFormat(rate); - - // Create audio file if path provided - if (filePath != NULL) { - CFURLRef fileURL = CFURLCreateFromFileSystemRepresentation( - NULL, (const UInt8 *)filePath, strlen(filePath), false); - if (fileURL == NULL) return -2; - - OSStatus status = AudioFileCreateWithURL(fileURL, kAudioFileWAVEType, - &audioFormat, kAudioFileFlags_EraseFile, &audioFile); - CFRelease(fileURL); - if (status != noErr) return -3; +static Capture *startCapture(const char *filePath, double rate, uintptr_t handle, OSStatus *status) { + Capture *capture = calloc(1, sizeof(Capture)); + if (capture == NULL) { + *status = ENOMEM; + return NULL; } - - // Create audio queue - OSStatus status = AudioQueueNewInput(&audioFormat, AudioInputCallback, NULL, - NULL, kCFRunLoopCommonModes, 0, &audioQueue); - if (status != noErr) { - if (audioFile) { AudioFileClose(audioFile); audioFile = NULL; } - return -4; + int mutexStatus = pthread_mutex_init(&capture->mutex, NULL); + if (mutexStatus != 0) { + *status = mutexStatus; + free(capture); + return NULL; } + capture->handle = handle; + capture->format.mSampleRate = rate; + capture->format.mFormatID = kAudioFormatLinearPCM; + capture->format.mFormatFlags = kLinearPCMFormatFlagIsSignedInteger | kLinearPCMFormatFlagIsPacked; + capture->format.mFramesPerPacket = 1; + capture->format.mChannelsPerFrame = CHANNELS; + capture->format.mBitsPerChannel = BITS_PER_SAMPLE; + capture->format.mBytesPerPacket = (BITS_PER_SAMPLE / 8) * CHANNELS; + capture->format.mBytesPerFrame = (BITS_PER_SAMPLE / 8) * CHANNELS; - // Allocate and enqueue buffers - for (int i = 0; i < NUM_BUFFERS; i++) { - status = AudioQueueAllocateBuffer(audioQueue, bufferSize, &buffers[i]); - if (status != noErr) { - AudioQueueDispose(audioQueue, true); - if (audioFile) { AudioFileClose(audioFile); audioFile = NULL; } - return -5; + if (filePath != NULL) { + CFURLRef url = CFURLCreateFromFileSystemRepresentation( + NULL, (const UInt8 *)filePath, strlen(filePath), false); + if (url == NULL) { + *status = EINVAL; + goto fail; } - AudioQueueEnqueueBuffer(audioQueue, buffers[i], 0, NULL); + *status = AudioFileCreateWithURL(url, kAudioFileWAVEType, + &capture->format, kAudioFileFlags_EraseFile, &capture->file); + CFRelease(url); + if (*status != noErr) goto fail; } - currentPacket = 0; - isCapturing = 1; + *status = AudioQueueNewInput(&capture->format, AudioInputCallback, capture, + NULL, kCFRunLoopCommonModes, 0, &capture->queue); + if (*status != noErr) goto fail; - status = AudioQueueStart(audioQueue, NULL); - if (status != noErr) { - isCapturing = 0; - AudioQueueDispose(audioQueue, true); - if (audioFile) { AudioFileClose(audioFile); audioFile = NULL; } - return -6; + UInt32 bufferSize = (UInt32)(rate * 0.1) * (BITS_PER_SAMPLE / 8) * CHANNELS; + for (int i = 0; i < NUM_BUFFERS; i++) { + AudioQueueBufferRef buffer; + *status = AudioQueueAllocateBuffer(capture->queue, bufferSize, &buffer); + if (*status != noErr) goto fail; + *status = AudioQueueEnqueueBuffer(capture->queue, buffer, 0, NULL); + if (*status != noErr) goto fail; } - - return 0; -} - -// Stop capturing -static int stopCapture() { - if (!isCapturing) return -1; - - isCapturing = 0; - AudioQueueStop(audioQueue, true); - AudioQueueDispose(audioQueue, true); - audioQueue = NULL; - - if (audioFile != NULL) { - AudioFileClose(audioFile); - audioFile = NULL; - } - - return 0; + *status = AudioQueueStart(capture->queue, NULL); + if (*status != noErr) goto fail; + capture->started = true; + return capture; + +fail: + freeCapture(capture); + return NULL; } */ import "C" import ( + "context" "errors" - "sync" - "time" + "fmt" + "runtime/cgo" + "strings" "unsafe" ) -// Common sample rates -const ( - SampleRate44100 = 44100 // CD quality, used for WAV recording - SampleRate24000 = 24000 // Required by OpenAI Realtime API -) - -// AudioHandler is called for each chunk of captured audio data. -// The data is mono, 16-bit signed PCM (little-endian). -type AudioHandler func(data []byte) - -// Capturer records audio from the microphone using CoreAudio. -type Capturer struct { - mu sync.Mutex - capturing bool - handler AudioHandler - sampleRate int +type callbacks struct { + handler AudioHandler + errors chan int } -// Global instance for C callback -var ( - globalCapturer *Capturer - globalCapturerMu sync.Mutex -) - //export goAudioCallback -func goAudioCallback(data unsafe.Pointer, size C.int) { - globalCapturerMu.Lock() - c := globalCapturer - globalCapturerMu.Unlock() - - if c != nil && c.handler != nil { - c.handler(C.GoBytes(data, size)) +func goAudioCallback(handle uintptr, data unsafe.Pointer, size, status int32) { + cb := cgo.Handle(handle).Value().(*callbacks) + if status != 0 { + select { + case cb.errors <- int(status): + default: + } + return + } + if cb.handler != nil { + cb.handler(C.GoBytes(data, C.int(size))) } } -// NewCapturer creates a new audio capturer with the specified sample rate. -func NewCapturer(sampleRate int) *Capturer { - return &Capturer{sampleRate: sampleRate} -} - -// Start begins capturing audio. If filePath is non-empty, audio is also saved to a WAV file. -// The handler is called for each chunk of audio data (can be nil if only recording to file). -func (c *Capturer) Start(filePath string, handler AudioHandler) error { - c.mu.Lock() - defer c.mu.Unlock() - - if c.capturing { - return errors.New("already capturing") +// Run captures until ctx is canceled. A non-empty filePath also records a WAV file. +// Each invocation is independent. The optional ready callback runs after capture starts. +// Handlers run synchronously; Run waits for them before returning, so they must not wait for Run. +// Canceling ctx normally returns nil. Native capture and cleanup failures return an error. +func (c *Capturer) Run(ctx context.Context, filePath string, handler AudioHandler, ready func()) (err error) { + if ctx.Err() != nil { + return nil } - - c.handler = handler - - globalCapturerMu.Lock() - globalCapturer = c - globalCapturerMu.Unlock() - - var cPath *C.char - if filePath != "" { - cPath = C.CString(filePath) - defer C.free(unsafe.Pointer(cPath)) + // C.GoBytes takes a signed 32-bit size for the 100 ms PCM buffer. + if c.sampleRate < 10 || uint64(c.sampleRate/10)*2 > 1<<31-1 { + return fmt.Errorf("invalid audio sample rate: %d", c.sampleRate) } - - if result := C.startCapture(cPath, C.double(c.sampleRate)); result != 0 { - return errorFromCode(int(result)) + if strings.IndexByte(filePath, 0) >= 0 { + return errors.New("audio file path contains a NUL byte") } - c.capturing = true - return nil -} - -// Stop ends audio capture. -func (c *Capturer) Stop() error { - c.mu.Lock() - defer c.mu.Unlock() + cb := &callbacks{handler: handler, errors: make(chan int, 1)} + handle := cgo.NewHandle(cb) + defer handle.Delete() - if !c.capturing { - return errors.New("not capturing") - } - - result := C.stopCapture() - if result != 0 { - return errorFromCode(int(result)) + var path *C.char + if filePath != "" { + path = C.CString(filePath) + defer C.free(unsafe.Pointer(path)) } - - c.capturing = false - - globalCapturerMu.Lock() - defer globalCapturerMu.Unlock() - globalCapturer = nil - - return nil -} - -// IsCapturing returns true if currently capturing. -func (c *Capturer) IsCapturing() bool { - c.mu.Lock() - defer c.mu.Unlock() - return c.capturing -} - -// Record captures audio for the specified duration and saves to the given WAV file. -func (c *Capturer) Record(filePath string, duration time.Duration) error { - err := c.Start(filePath, nil) - if err != nil { - return err + var status C.OSStatus + capture := C.startCapture(path, C.double(c.sampleRate), C.uintptr_t(handle), &status) + if capture == nil { + return fmt.Errorf("start audio capture: OSStatus %d", status) } - time.Sleep(duration) - return c.Stop() -} - -// Stream captures audio for the specified duration, calling handler for each chunk. -func (c *Capturer) Stream(duration time.Duration, handler AudioHandler) error { - if err := c.Start("", handler); err != nil { - return err + defer func() { + // Native callbacks must finish before their Go handle is deleted. + if status := C.freeCapture(capture); status != 0 { + err = fmt.Errorf("audio capture: OSStatus %d", status) + } + }() + + if ready != nil { + ready() } - time.Sleep(duration) - return c.Stop() -} - -var errorMessages = map[int]string{ - -1: "already capturing", - -2: "failed to create file URL", - -3: "failed to create audio file", - -4: "failed to create audio queue", - -5: "failed to allocate audio buffers", - -6: "failed to start audio queue", -} - -func errorFromCode(code int) error { - if msg, ok := errorMessages[code]; ok { - return errors.New(msg) + select { + case <-ctx.Done(): + case <-cb.errors: } - return errors.New("unknown error") + return nil } diff --git a/pkg/audio/capture/capture_darwin_test.go b/pkg/audio/capture/capture_darwin_test.go new file mode 100644 index 0000000000..4b9e816b64 --- /dev/null +++ b/pkg/audio/capture/capture_darwin_test.go @@ -0,0 +1,150 @@ +//go:build darwin && cgo && !no_audio + +package capture + +import ( + "context" + "os" + "path/filepath" + "runtime/cgo" + "sync" + "testing" + "unsafe" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRunCanceledBeforeStart(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + path := filepath.Join(t.TempDir(), "unused.wav") + capturer := NewCapturer(SampleRate24000) + for range 2 { + err := capturer.Run(ctx, path, func([]byte) { + t.Error("unexpected audio callback") + }, func() { + t.Error("unexpected readiness callback") + }) + require.NoError(t, err) + } + _, err := os.Stat(path) + require.ErrorIs(t, err, os.ErrNotExist) +} + +func TestRunInvalidInput(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + sampleRate int + path string + err string + }{ + {name: "zero rate", err: "invalid audio sample rate"}, + {name: "negative rate", sampleRate: -1, err: "invalid audio sample rate"}, + {name: "empty buffer", sampleRate: 9, err: "invalid audio sample rate"}, + {name: "oversized buffer", sampleRate: 1 << 40, err: "invalid audio sample rate"}, + {name: "NUL path", sampleRate: SampleRate24000, path: "audio\x00.wav", err: "NUL byte"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + err := NewCapturer(tt.sampleRate).Run(t.Context(), tt.path, nil, func() { + t.Error("unexpected readiness callback") + }) + require.ErrorContains(t, err, tt.err) + }) + } +} + +func TestRunFileCreationFailure(t *testing.T) { + t.Parallel() + + capturer := NewCapturer(SampleRate24000) + // File creation precedes queue creation, so this cannot access the microphone. + path := filepath.Join(t.TempDir(), "missing", "audio.wav") + for range 4 { + t.Run("run", func(t *testing.T) { + t.Parallel() + err := capturer.Run(t.Context(), path, func([]byte) { + t.Error("unexpected audio callback") + }, func() { + t.Error("unexpected readiness callback") + }) + require.ErrorContains(t, err, "start audio capture") + }) + } +} + +func TestAudioCallbackCopiesData(t *testing.T) { + t.Parallel() + + var received []byte + handle := cgo.NewHandle(&callbacks{handler: func(data []byte) { received = data }}) + defer handle.Delete() + + data := []byte{1, 2, 3, 4} + goAudioCallback(uintptr(handle), unsafe.Pointer(&data[0]), int32(len(data)), 0) + data[0] = 9 + assert.Equal(t, []byte{1, 2, 3, 4}, received) +} + +func TestAudioCallbacksAreIndependent(t *testing.T) { + t.Parallel() + + entered := make(chan struct{}) + release := make(chan struct{}) + done := make(chan struct{}) + var unblock sync.Once + defer unblock.Do(func() { close(release) }) + + first := cgo.NewHandle(&callbacks{handler: func([]byte) { + close(entered) + <-release + }}) + defer first.Delete() + go func() { + defer close(done) + goAudioCallback(uintptr(first), nil, 0, 0) + }() + <-entered + + called := false + second := cgo.NewHandle(&callbacks{handler: func([]byte) { called = true }}) + goAudioCallback(uintptr(second), nil, 0, 0) + second.Delete() + assert.True(t, called) + select { + case <-done: + t.Error("first callback returned before its handler finished") + default: + } + unblock.Do(func() { close(release) }) + <-done +} + +func TestAudioCallbackReportsError(t *testing.T) { + t.Parallel() + + cb := &callbacks{ + handler: func([]byte) { t.Error("unexpected audio callback") }, + errors: make(chan int, 1), + } + handle := cgo.NewHandle(cb) + defer handle.Delete() + + goAudioCallback(uintptr(handle), nil, 0, -1) + goAudioCallback(uintptr(handle), nil, 0, -2) + require.Equal(t, -1, <-cb.errors) +} + +func TestAudioCallbackWithoutHandler(t *testing.T) { + t.Parallel() + + handle := cgo.NewHandle(&callbacks{}) + defer handle.Delete() + goAudioCallback(uintptr(handle), nil, 0, 0) +} diff --git a/pkg/audio/capture/capture_other.go b/pkg/audio/capture/capture_other.go index 3ce680336e..796991283b 100644 --- a/pkg/audio/capture/capture_other.go +++ b/pkg/audio/capture/capture_other.go @@ -1,59 +1,10 @@ -//go:build !darwin || no_audio +//go:build !darwin || !cgo || no_audio -// Package capture provides audio capture functionality. -// This is a stub implementation for non-macOS platforms. package capture -import ( - "errors" - "time" -) +import "context" -// Common sample rates -const ( - SampleRate44100 = 44100 // CD quality, used for WAV recording - SampleRate24000 = 24000 // Required by OpenAI Realtime API -) - -// ErrNotSupported is returned when audio capture is not supported on the current platform. -var ErrNotSupported = errors.New("audio capture is only supported on macOS") - -// AudioHandler is called for each chunk of captured audio data. -// The data is mono, 16-bit signed PCM (little-endian). -type AudioHandler func(data []byte) - -// Capturer records audio from the microphone. -// On non-macOS platforms, this is a stub that returns errors. -type Capturer struct { - sampleRate int -} - -// NewCapturer creates a new audio capturer with the specified sample rate. -func NewCapturer(sampleRate int) *Capturer { - return &Capturer{sampleRate: sampleRate} -} - -// Start returns ErrNotSupported on non-macOS platforms. -func (c *Capturer) Start(filePath string, handler AudioHandler) error { - return ErrNotSupported -} - -// Stop returns ErrNotSupported on non-macOS platforms. -func (c *Capturer) Stop() error { - return ErrNotSupported -} - -// IsCapturing always returns false on non-macOS platforms. -func (c *Capturer) IsCapturing() bool { - return false -} - -// Record returns ErrNotSupported on non-macOS platforms. -func (c *Capturer) Record(filePath string, duration time.Duration) error { - return ErrNotSupported -} - -// Stream returns ErrNotSupported on non-macOS platforms. -func (c *Capturer) Stream(duration time.Duration, handler AudioHandler) error { +// Run returns ErrNotSupported when native audio capture is unavailable. +func (c *Capturer) Run(context.Context, string, AudioHandler, func()) error { return ErrNotSupported } diff --git a/pkg/audio/capture/capture_other_test.go b/pkg/audio/capture/capture_other_test.go new file mode 100644 index 0000000000..f0dfe49cb8 --- /dev/null +++ b/pkg/audio/capture/capture_other_test.go @@ -0,0 +1,29 @@ +//go:build !darwin || !cgo || no_audio + +package capture + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRunNotSupported(t *testing.T) { + t.Parallel() + + capturer := NewCapturer(SampleRate24000) + for _, canceled := range []bool{false, true} { + ctx, cancel := context.WithCancel(t.Context()) + if canceled { + cancel() + } + err := capturer.Run(ctx, "unused.wav", func([]byte) { + t.Error("unexpected audio callback") + }, func() { + t.Error("unexpected readiness callback") + }) + cancel() + require.ErrorIs(t, err, ErrNotSupported) + } +} diff --git a/pkg/audio/transcribe/finish_test.go b/pkg/audio/transcribe/finish_test.go new file mode 100644 index 0000000000..1bb9cd8001 --- /dev/null +++ b/pkg/audio/transcribe/finish_test.go @@ -0,0 +1,503 @@ +package transcribe + +import ( + "context" + "encoding/base64" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/audio/capture" +) + +func readClientEvent(t *testing.T, conn *websocket.Conn, want string) map[string]any { + t.Helper() + var event map[string]any + if err := conn.ReadJSON(&event); err != nil { + t.Error(err) + return nil + } + assert.Equal(t, want, event["type"]) + return event +} + +func writeServerEvent(t *testing.T, conn *websocket.Conn, event any) { + t.Helper() + assert.NoError(t, conn.WriteJSON(event)) +} + +func acknowledgeFinish(t *testing.T, conn *websocket.Conn) { + t.Helper() + update := readClientEvent(t, conn, "session.update") + assert.Equal(t, map[string]any{"turn_detection": nil}, update["session"]) + writeServerEvent(t, conn, map[string]any{ + "type": "session.updated", "session": map[string]any{"turn_detection": nil}, + }) + commit := readClientEvent(t, conn, "input_audio_buffer.commit") + assert.Equal(t, finalCommitID, commit["event_id"]) +} + +func commitItem(t *testing.T, conn *websocket.Conn, id string) { + t.Helper() + writeServerEvent(t, conn, map[string]string{"type": "input_audio_buffer.committed", "item_id": id}) +} + +func completeItem(t *testing.T, conn *websocket.Conn, id, text string) { + t.Helper() + writeServerEvent(t, conn, map[string]string{ + "type": "conversation.item.input_audio_transcription.completed", "item_id": id, "transcript": text, + }) +} + +func emptyFinalCommit(t *testing.T, conn *websocket.Conn) { + t.Helper() + writeServerEvent(t, conn, map[string]any{ + "type": "error", "error": map[string]string{ + "code": "input_audio_buffer_commit_empty", "message": "buffer empty", "event_id": finalCommitID, + }, + }) +} + +func TestRun_FinishDrainsAndCommits(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + for _, text := range []string{"first", "second", "cleanup tail"} { + event := readClientEvent(t, conn, "input_audio_buffer.append") + assert.Equal(t, base64.StdEncoding.EncodeToString([]byte(text)), event["audio"]) + } + acknowledgeFinish(t, conn) + commitItem(t, conn, "final") + completeItem(t, conn, "final", "all captured audio") + _, _, _ = conn.ReadMessage() + }) + var cleaned atomic.Bool + transcriber.capture = captureFunc(func(ctx context.Context, _ string, handler capture.AudioHandler, ready func()) error { + defer cleaned.Store(true) + handler([]byte("first")) + handler([]byte("second")) + ready() + <-ctx.Done() + handler([]byte("cleanup tail")) + return nil + }) + var texts []string + err := transcriber.Run(t.Context(), finish, func(ctx context.Context, text string) { + if err := ctx.Err(); err != nil { + t.Errorf("finish canceled the transcript handler: %v", err) + } + texts = append(texts, text) + }, func(context.Context) { close(finish) }) + require.NoError(t, err) + assert.True(t, cleaned.Load()) + assert.Equal(t, []string{"all captured audio"}, texts) +} + +func TestRun_FinishWaitsForVADItemsInOrder(t *testing.T) { + t.Parallel() + for _, empty := range []bool{false, true} { + t.Run(map[bool]string{false: "final audio", true: "empty final commit"}[empty], func(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + completed := make(chan struct{}) + release := make(chan struct{}) + defer close(release) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + commitItem(t, conn, "first-vad") + writeServerEvent(t, conn, map[string]string{ + "type": "conversation.item.input_audio_transcription.delta", "item_id": "first-vad", "delta": "first", + }) + close(finish) + update := readClientEvent(t, conn, "session.update") + assert.Equal(t, map[string]any{"turn_detection": nil}, update["session"]) + // A VAD commit already in flight is still before the disable acknowledgement. + commitItem(t, conn, "last-vad") + writeServerEvent(t, conn, map[string]any{ + "type": "session.updated", "session": map[string]any{"turn_detection": nil}, + }) + readClientEvent(t, conn, "input_audio_buffer.commit") + if empty { + emptyFinalCommit(t, conn) + } else { + commitItem(t, conn, "final") + completeItem(t, conn, "final", "third") + } + completeItem(t, conn, "last-vad", "second") + close(completed) + <-release + completeItem(t, conn, "first-vad", "first") + _, _, _ = conn.ReadMessage() + }) + var texts []string + done := make(chan error, 1) + go func() { + done <- transcriber.Run(t.Context(), finish, func(_ context.Context, text string) { + texts = append(texts, text) + }, nil) + }() + select { + case <-completed: + case <-time.After(5 * time.Second): + t.Fatal("server did not finish later items") + } + select { + case err := <-done: + t.Fatalf("returned before first VAD item completed: %v", err) + default: + } + release <- struct{}{} + require.NoError(t, waitRun(t, done)) + want := []string{"first", " second"} + if !empty { + want = append(want, " third") + } + assert.Equal(t, want, texts, "delta and completed text must not be duplicated") + }) + } +} + +func TestRun_FinishEmptyWithoutItems(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + acknowledgeFinish(t, conn) + emptyFinalCommit(t, conn) + _, _, _ = conn.ReadMessage() + }) + require.NoError(t, transcriber.Run(t.Context(), finish, nil, func(context.Context) { close(finish) })) +} + +func TestRun_EmptyCommitMustMatchFinalRequest(t *testing.T) { + t.Parallel() + for _, finishFirst := range []bool{false, true} { + t.Run(map[bool]string{false: "before finish", true: "different request"}[finishFirst], func(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + if finishFirst { + close(finish) + acknowledgeFinish(t, conn) + } + writeServerEvent(t, conn, map[string]any{ + "type": "error", "error": map[string]string{ + "code": "input_audio_buffer_commit_empty", "event_id": "unrelated", + }, + }) + _, _, _ = conn.ReadMessage() + }) + require.ErrorContains(t, transcriber.Run(t.Context(), finish, nil, nil), "input_audio_buffer_commit_empty") + }) + } +} + +func TestRun_TranscriptionFailure(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + close(finish) + acknowledgeFinish(t, conn) + commitItem(t, conn, "failed-item") + writeServerEvent(t, conn, map[string]any{ + "type": "conversation.item.input_audio_transcription.failed", "item_id": "failed-item", + "error": map[string]string{"code": "transcription_failed", "message": "cannot transcribe"}, + }) + _, _, _ = conn.ReadMessage() + }) + err := transcriber.Run(t.Context(), finish, nil, nil) + require.ErrorContains(t, err, "failed-item") + require.ErrorContains(t, err, "cannot transcribe") +} + +func TestRun_FinishBeforeStart(t *testing.T) { + t.Parallel() + transcriber := New("") + transcriber.supported = true + finish := make(chan struct{}) + close(finish) + require.NoError(t, transcriber.Run(t.Context(), finish, nil, func(context.Context) { t.Error("finished run became ready") })) +} + +func TestRun_FinishDuringUpgrade(t *testing.T) { + t.Parallel() + entered := make(chan struct{}) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + close(entered) + <-release + })) + t.Cleanup(server.Close) + defer close(release) + transcriber := New("test-api-key") + transcriber.supported = true + transcriber.endpoint = "ws" + strings.TrimPrefix(server.URL, "http") + transcriber.capture = captureFunc(func(context.Context, string, capture.AudioHandler, func()) error { + t.Error("finished upgrade started capture") + return nil + }) + finish := make(chan struct{}) + done := make(chan error, 1) + go func() { done <- transcriber.Run(t.Context(), finish, nil, nil) }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("upgrade request not received") + } + close(finish) + require.NoError(t, waitRun(t, done)) +} + +func TestRun_FinishUnblocksReady(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + acknowledgeFinish(t, conn) + emptyFinalCommit(t, conn) + _, _, _ = conn.ReadMessage() + }) + require.NoError(t, transcriber.Run(t.Context(), finish, nil, func(ctx context.Context) { + close(finish) + <-ctx.Done() + })) +} + +func TestRun_AbortDuringFinish(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + committed := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + acknowledgeFinish(t, conn) + close(committed) + _, _, _ = conn.ReadMessage() + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { done <- transcriber.Run(ctx, finish, nil, func(context.Context) { close(finish) }) }() + select { + case <-committed: + case <-time.After(5 * time.Second): + t.Fatal("final commit not sent") + } + cancel() + require.NoError(t, waitRun(t, done)) +} + +func TestRun_PreservesCaptureCleanupError(t *testing.T) { + t.Parallel() + for _, cause := range []string{"abort", "server error", "server disconnect", "finish"} { + t.Run(cause, func(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + finish := make(chan struct{}) + capturing := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + <-capturing + if cause == "server disconnect" { + return + } + if cause == "server error" { + writeServerEvent(t, conn, map[string]any{ + "type": "error", "error": map[string]string{"code": "denied", "message": "not allowed"}, + }) + } + _, _, _ = conn.ReadMessage() + }) + failure := errors.New("native cleanup failed") + transcriber.capture = captureFunc(func(ctx context.Context, _ string, _ capture.AudioHandler, ready func()) error { + close(capturing) + ready() + <-ctx.Done() + return failure + }) + err := transcriber.Run(ctx, finish, nil, func(context.Context) { + switch cause { + case "abort": + cancel() + case "finish": + close(finish) + } + }) + require.ErrorIs(t, err, failure) + if cause == "server error" { + require.ErrorContains(t, err, "not allowed") + } + if cause == "server disconnect" { + require.ErrorContains(t, err, "read transcription") + } + }) + } +} + +func TestRun_FinishTimeout(t *testing.T) { + t.Parallel() + for _, blocked := range []string{"VAD acknowledgement", "commit acknowledgement", "transcript", "handler", "write"} { + t.Run(blocked, func(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + release := make(chan struct{}) + defer close(release) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + switch blocked { + case "VAD acknowledgement": + readClientEvent(t, conn, "session.update") + case "commit acknowledgement": + acknowledgeFinish(t, conn) + case "transcript", "handler": + acknowledgeFinish(t, conn) + commitItem(t, conn, "final") + if blocked == "handler" { + completeItem(t, conn, "final", "hello") + } + } + <-release + }) + transcriber.finishTimeout = time.Second + if blocked == "write" { + transcriber.capture = captureFunc(func(ctx context.Context, _ string, handler capture.AudioHandler, ready func()) error { + ready() + handler(make([]byte, 16<<20)) + <-ctx.Done() + return nil + }) + } + done := make(chan error, 1) + go func() { + done <- transcriber.Run(t.Context(), finish, func(ctx context.Context, _ string) { <-ctx.Done() }, func(context.Context) { close(finish) }) + }() + require.ErrorIs(t, waitRun(t, done), context.DeadlineExceeded) + }) + } +} + +func TestRun_AbortSkipsFinalization(t *testing.T) { + t.Parallel() + peerClosed := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + defer close(peerClosed) + _, data, err := conn.ReadMessage() + if err == nil { + t.Error("abort did not close the connection") + } + assert.Empty(t, data, "abort must not send a final update or commit") + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + require.NoError(t, transcriber.Run(ctx, nil, nil, func(context.Context) { cancel() })) + select { + case <-peerClosed: + case <-time.After(5 * time.Second): + t.Fatal("abort did not close the connection") + } +} + +func TestRun_CompletedItemsDeliveredOnce(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + commitItem(t, conn, "first") + completeItem(t, conn, "first", "one") + completeItem(t, conn, "first", "one") + commitItem(t, conn, "first") + completeItem(t, conn, "second", "two") + commitItem(t, conn, "second") + close(finish) + acknowledgeFinish(t, conn) + emptyFinalCommit(t, conn) + _, _, _ = conn.ReadMessage() + }) + var texts []string + require.NoError(t, transcriber.Run(t.Context(), finish, func(_ context.Context, text string) { + texts = append(texts, text) + }, nil)) + assert.Equal(t, []string{"one", " two"}, texts) +} + +func TestRun_UnexpectedCaptureStop(t *testing.T) { + t.Parallel() + transcriber := testTranscriber(t, func(conn *websocket.Conn) { _, _, _ = conn.ReadMessage() }) + transcriber.capture = captureFunc(func(context.Context, string, capture.AudioHandler, func()) error { + return nil + }) + require.ErrorContains(t, transcriber.Run(t.Context(), nil, nil, nil), "audio capture stopped unexpectedly") +} + +func announceVAD(t *testing.T, conn *websocket.Conn, id string) { + t.Helper() + writeServerEvent(t, conn, map[string]string{"type": "input_audio_buffer.speech_started", "item_id": id}) + writeServerEvent(t, conn, map[string]string{"type": "input_audio_buffer.speech_stopped", "item_id": id}) +} + +func TestRun_FinishWaitsForLateVAD(t *testing.T) { + t.Parallel() + for _, empty := range []bool{false, true} { + t.Run(map[bool]string{false: "manual final", true: "empty final"}[empty], func(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + announceVAD(t, conn, "late-vad") + close(finish) + acknowledgeFinish(t, conn) + if empty { + emptyFinalCommit(t, conn) + } + commitItem(t, conn, "late-vad") + completeItem(t, conn, "late-vad", "first") + if !empty { + commitItem(t, conn, "manual-final") + completeItem(t, conn, "manual-final", "second") + } + _, _, _ = conn.ReadMessage() + }) + var texts []string + require.NoError(t, transcriber.Run(t.Context(), finish, func(_ context.Context, text string) { + texts = append(texts, text) + }, nil)) + want := []string{"first"} + if !empty { + want = append(want, " second") + } + assert.Equal(t, want, texts) + }) + } +} + +func TestRun_ManualCommitOfActiveSpeech(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + writeServerEvent(t, conn, map[string]string{"type": "input_audio_buffer.speech_started", "item_id": "active"}) + close(finish) + acknowledgeFinish(t, conn) + commitItem(t, conn, "active") + completeItem(t, conn, "active", "last words") + _, _, _ = conn.ReadMessage() + }) + var text string + require.NoError(t, transcriber.Run(t.Context(), finish, func(_ context.Context, delta string) { text += delta }, nil)) + assert.Equal(t, "last words", text) +} + +func TestRun_ManualCommitReplacesProvisionalVADItem(t *testing.T) { + t.Parallel() + finish := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + writeServerEvent(t, conn, map[string]string{"type": "input_audio_buffer.speech_started", "item_id": "provisional"}) + close(finish) + acknowledgeFinish(t, conn) + commitItem(t, conn, "manual-final") + completeItem(t, conn, "manual-final", "last words") + _, _, _ = conn.ReadMessage() + }) + var text string + require.NoError(t, transcriber.Run(t.Context(), finish, func(_ context.Context, delta string) { text += delta }, nil)) + assert.Equal(t, "last words", text) +} diff --git a/pkg/audio/transcribe/transcribe.go b/pkg/audio/transcribe/transcribe.go new file mode 100644 index 0000000000..1d1fb21d7b --- /dev/null +++ b/pkg/audio/transcribe/transcribe.go @@ -0,0 +1,390 @@ +// Package transcribe provides real-time audio transcription using OpenAI's Realtime API. +package transcribe + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "sync" + "time" + "unicode" + "unicode/utf8" + + "github.com/gorilla/websocket" + + "github.com/docker/docker-agent/pkg/audio/capture" +) + +const ( + openAIRealtimeURL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview" + finalCommitID = "transcribe-final-commit" +) + +// ErrNotSupported is returned when transcription is unavailable on this platform. +var ErrNotSupported = errors.New("speech-to-text is only supported on macOS") + +// TranscriptHandler receives completed transcripts in audio commit order. +// It must return when ctx is canceled. +type TranscriptHandler func(ctx context.Context, delta string) + +type audioCapture interface { + Run(ctx context.Context, filePath string, handler capture.AudioHandler, ready func()) error +} + +// Transcriber holds configuration; each Run owns its connection and audio capture. +type Transcriber struct { + apiKey string + endpoint string + capture audioCapture + supported bool + finishTimeout time.Duration +} + +// New creates a transcriber without opening a connection or microphone. +func New(apiKey string) *Transcriber { + return &Transcriber{ + apiKey: apiKey, + endpoint: openAIRealtimeURL, + capture: capture.NewCapturer(capture.SampleRate24000), + supported: audioSupported, + } +} + +// IsSupported reports whether microphone capture is available on this platform. +func (t *Transcriber) IsSupported() bool { return t.supported } + +// Run aborts promptly on ctx cancellation. Closing finish stops capture, drains +// captured audio, and waits up to ten seconds for final transcription. A nil +// finish channel disables graceful finishing. Finishing before connection setup +// completes cancels setup without starting capture. +// +// Ready runs after capture starts, with a context canceled on finish or abort. +// Transcript handlers remain active during finish. Callbacks must honor their +// contexts; Run joins them and capture cleanup before returning. Cancellation +// itself returns nil, but capture, server and finalization errors are preserved. +func (t *Transcriber) Run(ctx context.Context, finish <-chan struct{}, handler TranscriptHandler, ready func(context.Context)) (err error) { + if !t.IsSupported() { + return ErrNotSupported + } + select { + case <-ctx.Done(): + return nil + case <-finish: + return nil + default: + } + if t.apiKey == "" { + return errors.New("/speak sends audio directly to OpenAI; set a separate OPENAI_API_KEY environment variable even when chat uses Docker's models gateway, which does not support /speak") + } + + runCtx, cancel := context.WithCancel(ctx) + captureCtx, stopCapture := context.WithCancel(runCtx) + finished := make(chan struct{}) + var finishErr error + go func() { + defer close(finished) + select { + case <-runCtx.Done(): + return + case <-finish: + } + stopCapture() + timeout := t.finishTimeout + if timeout <= 0 { + timeout = 10 * time.Second + } + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-runCtx.Done(): + case <-timer.C: + finishErr = fmt.Errorf("finish transcription: %w", context.DeadlineExceeded) + cancel() + } + }() + defer func() { + cancel() + stopCapture() + <-finished + err = errors.Join(err, finishErr) + }() + + conn, response, err := dial(captureCtx, t.endpoint, http.Header{ + "Authorization": []string{"Bearer " + t.apiKey}, + "OpenAI-Beta": []string{"realtime=v1"}, + }) + if err != nil { + if response != nil && response.Body != nil { + response.Body.Close() + } + if expectedClose(captureCtx, err) { + return nil + } + return fmt.Errorf("connect /speak directly to OpenAI Realtime API (not Docker's models gateway): %w", err) + } + defer conn.Close() + closed := make(chan struct{}) + stopClose := context.AfterFunc(runCtx, func() { + conn.Close() // Interrupt blocked reads and writes before joining workers. + close(closed) + }) + defer func() { + if !stopClose() { + <-closed + } + }() + if captureCtx.Err() != nil { + return nil + } + if err := conn.WriteJSON(map[string]any{ + "type": "session.update", + "session": map[string]any{ + "modalities": []string{"text"}, + "input_audio_format": "pcm16", + "input_audio_transcription": map[string]string{ + "model": "whisper-1", + }, + "turn_detection": map[string]any{"type": "server_vad"}, + }, + }); err != nil { + if expectedClose(runCtx, err) { + return nil + } + return fmt.Errorf("configure transcription: %w", err) + } + + var workers sync.WaitGroup + var errorMu sync.Mutex + var runErr error + fail := func(err error) { + if err != nil { + errorMu.Lock() + runErr = errors.Join(runErr, err) + errorMu.Unlock() + cancel() + } + } + // Never block the native audio callback on network I/O. + audio := make(chan []byte, 32) + var overflow sync.Once + workers.Go(func() { + defer close(audio) + err := t.capture.Run(captureCtx, "", func(data []byte) { + if runCtx.Err() != nil { + return + } + select { + case audio <- data: + default: + overflow.Do(func() { fail(errors.New("audio upload fell behind capture")) }) + } + }, func() { + if ready != nil && captureCtx.Err() == nil { + ready(captureCtx) + } + }) + if err != nil { + fail(fmt.Errorf("capture audio: %w", err)) + } else if captureCtx.Err() == nil { + fail(errors.New("audio capture stopped unexpectedly")) + } + }) + vadDisabled := make(chan struct{}) + workers.Go(func() { + fail(writeAudio(runCtx, conn, audio, vadDisabled)) + }) + workers.Go(func() { + defer cancel() + fail(readLoop(runCtx, conn, handler, vadDisabled)) + }) + workers.Wait() + return runErr +} + +// Only suppress I/O errors attributable to our own cancellation/connection close. +func expectedClose(ctx context.Context, err error) bool { + return ctx.Err() != nil && (errors.Is(err, net.ErrClosed) || errors.Is(err, ctx.Err())) +} + +// Gorilla applies handshake deadlines but does not close an established TCP +// connection when its context is canceled during the HTTP upgrade. +func dial(ctx context.Context, endpoint string, headers http.Header) (*websocket.Conn, *http.Response, error) { + dialer := *websocket.DefaultDialer + var stopClose func() bool + var closed chan struct{} + dialer.NetDialContext = func(dialCtx context.Context, network, address string) (net.Conn, error) { + conn, err := (&net.Dialer{}).DialContext(dialCtx, network, address) + if err != nil { + return nil, err + } + closed = make(chan struct{}) + stopClose = context.AfterFunc(ctx, func() { + conn.Close() + close(closed) + }) + return conn, nil + } + defer func() { + if stopClose != nil && !stopClose() { + <-closed + } + }() + return dialer.DialContext(ctx, endpoint, headers) +} + +func writeAudio(ctx context.Context, conn *websocket.Conn, audio <-chan []byte, vadDisabled <-chan struct{}) error { + write := func(event any) error { + if ctx.Err() != nil { + return nil + } + if err := conn.WriteJSON(event); err != nil && !expectedClose(ctx, err) { + return fmt.Errorf("upload audio: %w", err) + } + return nil + } + for { + select { + case <-ctx.Done(): + return nil + case data, ok := <-audio: + if !ok { + // Disable new VAD turns before requesting the manual final commit. + if err := write(map[string]any{ + "type": "session.update", "session": map[string]any{"turn_detection": nil}, + }); err != nil { + return err + } + select { + case <-ctx.Done(): + return nil + case <-vadDisabled: + } + return write(map[string]string{"type": "input_audio_buffer.commit", "event_id": finalCommitID}) + } + if err := write(map[string]string{ + "type": "input_audio_buffer.append", "audio": base64.StdEncoding.EncodeToString(data), + }); err != nil { + return err + } + } + } +} + +type serverEvent struct { + Type string `json:"type"` + ItemID string `json:"item_id"` + Transcript string `json:"transcript"` + Session *struct { + TurnDetection json.RawMessage `json:"turn_detection"` + } `json:"session"` + Error *struct { + Code string `json:"code"` + Message string `json:"message"` + EventID string `json:"event_id"` + } `json:"error"` +} + +func readLoop(ctx context.Context, conn *websocket.Conn, handler TranscriptHandler, vadDisabled chan<- struct{}) error { + type transcript struct { + text string + completed bool + committed bool + started bool + automatic bool + } + items := make(map[string]*transcript) + var pending []*transcript + var disabled, finalResolved bool + var last rune + for { + _, msg, err := conn.ReadMessage() + if err != nil { + if expectedClose(ctx, err) { + return nil + } + return fmt.Errorf("read transcription: %w", err) + } + var event serverEvent + if json.Unmarshal(msg, &event) != nil { + continue + } + switch event.Type { + case "session.updated": + if !disabled && event.Session != nil && string(event.Session.TurnDetection) == "null" { + disabled = true + close(vadDisabled) + } + case "input_audio_buffer.speech_started", "input_audio_buffer.speech_stopped", "input_audio_buffer.committed", "conversation.item.input_audio_transcription.completed": + item := items[event.ItemID] + if item == nil { + item = &transcript{} + items[event.ItemID] = item + } + switch event.Type { + case "input_audio_buffer.speech_started", "input_audio_buffer.speech_stopped": + item.started = true + if event.Type == "input_audio_buffer.speech_stopped" { + item.automatic = true + } + if !item.committed { + item.committed = true + pending = append(pending, item) + } + case "input_audio_buffer.committed": + if disabled && !item.automatic { + // A manual commit may replace a started-only VAD item with a new ID. + for i, previous := range pending { + if previous != item && previous.started && !previous.automatic && !previous.completed { + pending[i] = item + item.committed = true + break + } + } + } + if !item.committed { + item.committed = true + pending = append(pending, item) + } + if disabled && !item.automatic { + finalResolved = true + } + case "conversation.item.input_audio_transcription.completed": + item.text, item.completed = event.Transcript, true + } + case "conversation.item.input_audio_transcription.failed": + if event.Error != nil { + return fmt.Errorf("transcription failed for item %s (%s): %s", event.ItemID, event.Error.Code, event.Error.Message) + } + return fmt.Errorf("transcription failed for item %s", event.ItemID) + case "error": + if event.Error != nil { + if disabled && event.Error.Code == "input_audio_buffer_commit_empty" && event.Error.EventID == finalCommitID { + finalResolved = true + } else { + return fmt.Errorf("transcription server (%s): %s", event.Error.Code, event.Error.Message) + } + } + } + // Completed text is authoritative; streaming deltas can interleave across items. + for len(pending) > 0 && pending[0].completed { + if handler != nil && pending[0].text != "" && ctx.Err() == nil { + text := pending[0].text + first, _ := utf8.DecodeRuneInString(text) + if last != 0 && !unicode.IsSpace(last) && !unicode.IsSpace(first) { + text = " " + text + } + handler(ctx, text) + last, _ = utf8.DecodeLastRuneInString(text) + } + pending[0].text = "" + pending = pending[1:] + } + if finalResolved && len(pending) == 0 { + return nil + } + } +} diff --git a/pkg/audio/transcribe/transcribe_darwin.go b/pkg/audio/transcribe/transcribe_darwin.go index 870b5bfa70..7765becc81 100644 --- a/pkg/audio/transcribe/transcribe_darwin.go +++ b/pkg/audio/transcribe/transcribe_darwin.go @@ -1,192 +1,5 @@ -//go:build darwin && !no_audio +//go:build darwin && cgo && !no_audio -// Package transcribe provides real-time audio transcription using OpenAI's Realtime API. package transcribe -import ( - "context" - "encoding/base64" - "encoding/json" - "errors" - "fmt" - "net/http" - "sync" - "sync/atomic" - - "github.com/gorilla/websocket" - - "github.com/docker/docker-agent/pkg/audio/capture" -) - -const openAIRealtimeURL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview" - -// TranscriptHandler is called when new transcription text is received. -type TranscriptHandler func(delta string) - -// Transcriber provides real-time audio transcription using OpenAI's Realtime API. -type Transcriber struct { - apiKey string - conn *websocket.Conn - capture *capture.Capturer - running atomic.Bool - cancel context.CancelFunc - writeMu sync.Mutex -} - -// serverEvent represents events from the OpenAI Realtime API. -type serverEvent struct { - Type string `json:"type"` - Delta string `json:"delta,omitempty"` - Error *struct { - Code string `json:"code"` - Message string `json:"message"` - } `json:"error,omitempty"` -} - -// New creates a new Transcriber with the given OpenAI API key. -func New(apiKey string) *Transcriber { - return &Transcriber{ - apiKey: apiKey, - capture: capture.NewCapturer(capture.SampleRate24000), - } -} - -// Start begins audio capture and transcription. The handler is called for each -// transcription delta received. Returns an error if already running or if -// connection fails. Call Stop to end transcription. -func (t *Transcriber) Start(ctx context.Context, handler TranscriptHandler) error { - if t.apiKey == "" { - return errors.New("/speak sends audio directly to OpenAI; set a separate OPENAI_API_KEY environment variable even when chat uses Docker's models gateway, which does not support /speak") - } - - if wasRunning := t.running.Swap(true); wasRunning { - return errors.New("transcriber already running") - } - - ctx, cancel := context.WithCancel(ctx) - t.cancel = cancel - - // Connect to OpenAI Realtime API - conn, _, err := websocket.DefaultDialer.DialContext(ctx, openAIRealtimeURL, http.Header{ //nolint:bodyclose // websocket upgrade response - "Authorization": []string{"Bearer " + t.apiKey}, - "OpenAI-Beta": []string{"realtime=v1"}, - }) - if err != nil { - t.running.Store(false) - return fmt.Errorf("connect /speak directly to OpenAI Realtime API (not Docker's models gateway): %w", err) - } - t.conn = conn - - // Configure session for transcription only - if err := conn.WriteJSON(map[string]any{ - "type": "session.update", - "session": map[string]any{ - "modalities": []string{"text"}, - "input_audio_format": "pcm16", - "input_audio_transcription": map[string]string{ - "model": "whisper-1", - }, - "turn_detection": map[string]any{ - "type": "server_vad", - }, - }, - }); err != nil { - conn.Close() - t.running.Store(false) - return fmt.Errorf("configure session: %w", err) - } - - // Start reading WebSocket messages - go t.readLoop(ctx, handler) - - // Start audio capture - err = t.capture.Start("", func(data []byte) { - t.writeMu.Lock() - defer t.writeMu.Unlock() - - if t.conn != nil { - _ = t.conn.WriteJSON(map[string]string{ - "type": "input_audio_buffer.append", - "audio": base64.StdEncoding.EncodeToString(data), - }) - } - }) - if err != nil { - conn.Close() - t.running.Store(false) - return fmt.Errorf("start capture: %w", err) - } - - return nil -} - -// Stop ends the transcription session and releases resources. -func (t *Transcriber) Stop() { - if wasRunning := t.running.Swap(false); !wasRunning { - return - } - - // Cancel the context to stop the read loop - if t.cancel != nil { - t.cancel() - t.cancel = nil - } - - // Stop audio capture - _ = t.capture.Stop() - - // Commit any remaining audio buffer - t.writeMu.Lock() - defer t.writeMu.Unlock() - if t.conn != nil { - _ = t.conn.WriteJSON(map[string]string{"type": "input_audio_buffer.commit"}) - _ = t.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) - t.conn.Close() - t.conn = nil - } -} - -// IsRunning returns true if transcription is currently active. -func (t *Transcriber) IsRunning() bool { - return t.running.Load() -} - -// IsSupported returns true on macOS where audio capture is available. -func (t *Transcriber) IsSupported() bool { - return true -} - -// readLoop reads messages from the WebSocket and calls the handler for transcription deltas. -func (t *Transcriber) readLoop(ctx context.Context, handler TranscriptHandler) { - for { - select { - case <-ctx.Done(): - return - default: - } - - _, msg, err := t.conn.ReadMessage() - if err != nil { - // Connection closed or error - return - } - - var event serverEvent - if json.Unmarshal(msg, &event) != nil { - continue - } - - switch event.Type { - case "conversation.item.input_audio_transcription.delta": - if handler != nil && event.Delta != "" { - handler(event.Delta) - } - case "error": - // Ignore empty buffer commit errors - if event.Error != nil && event.Error.Code != "input_audio_buffer_commit_empty" { - // Log or handle error - return - } - } - } -} +const audioSupported = true diff --git a/pkg/audio/transcribe/transcribe_darwin_test.go b/pkg/audio/transcribe/transcribe_darwin_test.go deleted file mode 100644 index d2eb79fd61..0000000000 --- a/pkg/audio/transcribe/transcribe_darwin_test.go +++ /dev/null @@ -1,37 +0,0 @@ -//go:build darwin && !no_audio - -package transcribe - -import ( - "context" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestStart_MissingAPIKey(t *testing.T) { - t.Parallel() - - transcriber := New("") - err := transcriber.Start(t.Context(), nil) - - require.EqualError(t, err, "/speak sends audio directly to OpenAI; set a separate OPENAI_API_KEY environment variable even when chat uses Docker's models gateway, which does not support /speak") - assert.False(t, transcriber.IsRunning()) - assert.Nil(t, transcriber.conn) -} - -func TestStart_ConnectionFailure(t *testing.T) { - t.Parallel() - - ctx, cancel := context.WithCancel(t.Context()) - cancel() - - transcriber := New("test-api-key") - err := transcriber.Start(ctx, nil) - - require.ErrorIs(t, err, context.Canceled) - assert.Contains(t, err.Error(), "connect /speak directly to OpenAI Realtime API (not Docker's models gateway): ") - assert.False(t, transcriber.IsRunning()) - assert.Nil(t, transcriber.conn) -} diff --git a/pkg/audio/transcribe/transcribe_other.go b/pkg/audio/transcribe/transcribe_other.go index fa25aadcc2..6d07901e8b 100644 --- a/pkg/audio/transcribe/transcribe_other.go +++ b/pkg/audio/transcribe/transcribe_other.go @@ -1,43 +1,5 @@ -//go:build !darwin || no_audio +//go:build !darwin || !cgo || no_audio -// Package transcribe provides real-time audio transcription using OpenAI's Realtime API. -// This is a stub implementation for non-macOS platforms. package transcribe -import ( - "context" - "errors" -) - -// ErrNotSupported is returned when transcription is not supported on the current platform. -var ErrNotSupported = errors.New("speech-to-text is only supported on macOS") - -// TranscriptHandler is called when new transcription text is received. -type TranscriptHandler func(delta string) - -// Transcriber provides real-time audio transcription using OpenAI's Realtime API. -// On non-macOS platforms, this is a stub that returns errors. -type Transcriber struct{} - -// New creates a new Transcriber with the given OpenAI API key. -func New(apiKey string) *Transcriber { - return &Transcriber{} -} - -// Start returns ErrNotSupported on non-macOS platforms. -func (t *Transcriber) Start(ctx context.Context, handler TranscriptHandler) error { - return ErrNotSupported -} - -// Stop is a no-op on non-macOS platforms. -func (t *Transcriber) Stop() {} - -// IsRunning always returns false on non-macOS platforms. -func (t *Transcriber) IsRunning() bool { - return false -} - -// IsSupported returns false on non-macOS platforms. -func (t *Transcriber) IsSupported() bool { - return false -} +const audioSupported = false diff --git a/pkg/audio/transcribe/transcribe_test.go b/pkg/audio/transcribe/transcribe_test.go new file mode 100644 index 0000000000..3662f74abf --- /dev/null +++ b/pkg/audio/transcribe/transcribe_test.go @@ -0,0 +1,316 @@ +package transcribe + +import ( + "context" + "encoding/base64" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/audio/capture" +) + +type captureFunc func(context.Context, string, capture.AudioHandler, func()) error + +func (f captureFunc) Run(ctx context.Context, path string, handler capture.AudioHandler, ready func()) error { + return f(ctx, path, handler, ready) +} + +func testTranscriber(t *testing.T, handler func(*websocket.Conn)) *Transcriber { + t.Helper() + var connections sync.WaitGroup + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + connections.Add(1) + defer connections.Done() + assert.Equal(t, "Bearer test-api-key", r.Header.Get("Authorization")) + assert.Equal(t, "realtime=v1", r.Header.Get("OpenAI-Beta")) + conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if !assert.NoError(t, err) { + return + } + defer conn.Close() + var config map[string]any + if !assert.NoError(t, conn.ReadJSON(&config)) { + return + } + assert.Equal(t, "session.update", config["type"]) + handler(conn) + })) + t.Cleanup(func() { + server.Close() + connections.Wait() + }) + return &Transcriber{ + apiKey: "test-api-key", endpoint: "ws" + strings.TrimPrefix(server.URL, "http"), supported: true, + capture: captureFunc(func(ctx context.Context, _ string, _ capture.AudioHandler, ready func()) error { + if ready != nil { + ready() + } + <-ctx.Done() + return nil + }), + } +} + +func waitRun(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(5 * time.Second): + t.Fatal("transcription did not finish") + return nil + } +} + +func TestRun_MissingAPIKey(t *testing.T) { + t.Parallel() + transcriber := New("") + transcriber.supported = true + err := transcriber.Run(t.Context(), nil, nil, nil) + require.EqualError(t, err, "/speak sends audio directly to OpenAI; set a separate OPENAI_API_KEY environment variable even when chat uses Docker's models gateway, which does not support /speak") +} + +func TestRun_CanceledBeforeStart(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + transcriber := New("test-api-key") + transcriber.supported = true + require.NoError(t, transcriber.Run(ctx, nil, nil, func(context.Context) { t.Error("canceled run became ready") })) +} + +func TestRun_Unsupported(t *testing.T) { + t.Parallel() + transcriber := New("test-api-key") + transcriber.supported = false + require.ErrorIs(t, transcriber.Run(t.Context(), nil, nil, nil), ErrNotSupported) +} + +func TestRun_ConnectionFailure(t *testing.T) { + t.Parallel() + transcriber := New("test-api-key") + transcriber.supported = true + transcriber.endpoint = ":invalid" + err := transcriber.Run(t.Context(), nil, nil, nil) + require.ErrorContains(t, err, "connect /speak directly to OpenAI Realtime API") +} + +func TestRun_CaptureFailure(t *testing.T) { + t.Parallel() + transcriber := testTranscriber(t, func(conn *websocket.Conn) { _, _, _ = conn.ReadMessage() }) + failure := errors.New("microphone unavailable") + transcriber.capture = captureFunc(func(context.Context, string, capture.AudioHandler, func()) error { + return failure + }) + require.ErrorIs(t, transcriber.Run(t.Context(), nil, nil, func(context.Context) { t.Error("failed capture became ready") }), failure) +} + +func TestRun_AudioAndTranscript(t *testing.T) { + t.Parallel() + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + var audio map[string]string + if err := conn.ReadJSON(&audio); err != nil { + t.Error(err) + return + } + assert.Equal(t, "input_audio_buffer.append", audio["type"]) + assert.Equal(t, base64.StdEncoding.EncodeToString([]byte("audio")), audio["audio"]) + if err := conn.WriteJSON(map[string]string{"type": "input_audio_buffer.committed", "item_id": "item-1"}); err != nil { + t.Error(err) + return + } + if err := conn.WriteJSON(map[string]string{ + "type": "conversation.item.input_audio_transcription.completed", "item_id": "item-1", "transcript": "hello", + }); err != nil { + t.Error(err) + return + } + _, _, _ = conn.ReadMessage() + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var cleaned atomic.Bool + transcriber.capture = captureFunc(func(ctx context.Context, _ string, handler capture.AudioHandler, ready func()) error { + defer cleaned.Store(true) + ready() + handler([]byte("audio")) + <-ctx.Done() + return nil + }) + done := make(chan error, 1) + go func() { + done <- transcriber.Run(ctx, nil, func(_ context.Context, delta string) { + assert.Equal(t, "hello", delta) + cancel() + }, func(context.Context) {}) + }() + require.NoError(t, waitRun(t, done)) + assert.True(t, cleaned.Load(), "Run must join capture cleanup") +} + +func TestRun_ServerFailure(t *testing.T) { + t.Parallel() + for _, serverError := range []bool{false, true} { + t.Run(map[bool]string{false: "disconnect", true: "error event"}[serverError], func(t *testing.T) { + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + if serverError { + assert.NoError(t, conn.WriteJSON(map[string]any{ + "type": "error", "error": map[string]string{"code": "denied", "message": "not allowed"}, + })) + } + }) + var cleaned atomic.Bool + transcriber.capture = captureFunc(func(ctx context.Context, _ string, _ capture.AudioHandler, _ func()) error { + defer cleaned.Store(true) + <-ctx.Done() + return nil + }) + err := transcriber.Run(t.Context(), nil, nil, nil) + require.Error(t, err) + if serverError { + require.ErrorContains(t, err, "not allowed") + } + assert.True(t, cleaned.Load()) + }) + } +} + +func TestRun_ConcurrentRuns(t *testing.T) { + t.Parallel() + transcriber := testTranscriber(t, func(conn *websocket.Conn) { _, _, _ = conn.ReadMessage() }) + first, cancelFirst := context.WithCancel(t.Context()) + defer cancelFirst() + second, cancelSecond := context.WithCancel(t.Context()) + defer cancelSecond() + ready := make(chan struct{}, 2) + doneFirst, doneSecond := make(chan error, 1), make(chan error, 1) + go func() { doneFirst <- transcriber.Run(first, nil, nil, func(context.Context) { ready <- struct{}{} }) }() + go func() { doneSecond <- transcriber.Run(second, nil, nil, func(context.Context) { ready <- struct{}{} }) }() + for range 2 { + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("transcription did not become ready") + } + } + cancelFirst() + require.NoError(t, waitRun(t, doneFirst)) + select { + case err := <-doneSecond: + t.Fatalf("canceling first run ended second run: %v", err) + default: + } + cancelSecond() + require.NoError(t, waitRun(t, doneSecond)) +} + +func TestRun_CancelBlockedHandler(t *testing.T) { + t.Parallel() + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + if err := conn.WriteJSON(map[string]string{"type": "input_audio_buffer.committed", "item_id": "item-1"}); err != nil { + t.Error(err) + return + } + if err := conn.WriteJSON(map[string]string{ + "type": "conversation.item.input_audio_transcription.completed", "item_id": "item-1", "transcript": "hello", + }); err != nil { + t.Error(err) + return + } + _, _, _ = conn.ReadMessage() + }) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + entered := make(chan struct{}) + done := make(chan error, 1) + go func() { + done <- transcriber.Run(ctx, nil, func(ctx context.Context, _ string) { + close(entered) + <-ctx.Done() + }, nil) + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("handler not called") + } + cancel() + require.NoError(t, waitRun(t, done)) +} + +func TestRun_AudioOverflowCancelsBlockedWrite(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + defer close(release) + transcriber := testTranscriber(t, func(*websocket.Conn) { + <-release // Keep the connection open without draining audio. + }) + transcriber.capture = captureFunc(func(ctx context.Context, _ string, handler capture.AudioHandler, _ func()) error { + chunk := make([]byte, 1<<20) + for range 100 { + handler(chunk) + } + <-ctx.Done() + return nil + }) + done := make(chan error, 1) + go func() { done <- transcriber.Run(t.Context(), nil, nil, nil) }() + require.ErrorContains(t, waitRun(t, done), "audio upload fell behind capture") +} + +func TestRun_CancelDuringUpgrade(t *testing.T) { + t.Parallel() + entered := make(chan struct{}) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + close(entered) + <-release + })) + t.Cleanup(server.Close) + defer close(release) + transcriber := New("test-api-key") + transcriber.supported = true + transcriber.endpoint = "ws" + strings.TrimPrefix(server.URL, "http") + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { done <- transcriber.Run(ctx, nil, nil, nil) }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("upgrade request not received") + } + cancel() + require.NoError(t, waitRun(t, done)) +} + +func TestRun_ServerFailureCancelsBlockedReady(t *testing.T) { + t.Parallel() + ready := make(chan struct{}) + transcriber := testTranscriber(t, func(conn *websocket.Conn) { + <-ready + if err := conn.WriteJSON(map[string]any{ + "type": "error", "error": map[string]string{"code": "denied", "message": "not allowed"}, + }); err != nil { + t.Error(err) + } + }) + done := make(chan error, 1) + go func() { + done <- transcriber.Run(t.Context(), nil, nil, func(ctx context.Context) { + close(ready) + <-ctx.Done() + }) + }() + require.ErrorContains(t, waitRun(t, done), "not allowed") +} diff --git a/pkg/tui/handlers.go b/pkg/tui/handlers.go index ecb6b5c898..b6dfbd5d6d 100644 --- a/pkg/tui/handlers.go +++ b/pkg/tui/handlers.go @@ -1244,69 +1244,6 @@ func (m *appModel) handleAttachFile(filePath string) (tea.Model, tea.Cmd) { }) } -// --- Speech-to-text --- - -func (m *appModel) handleStartSpeak() (tea.Model, tea.Cmd) { - if m.transcriber.IsRunning() { - return m, nil - } - - // Close any previous channel to unblock stale waitForTranscript goroutines. - m.closeTranscriptCh() - - ch := make(chan string, 100) - m.transcriptCh = ch - err := m.transcriber.Start(m.ctx(), func(delta string) { - select { - case ch <- delta: - default: - } - }) - if err != nil { - m.closeTranscriptCh() - return m, notification.ErrorCmd(fmt.Sprintf("Failed to start listening: %v", err)) - } - - return m, tea.Batch( - notification.InfoCmd("🎤 Listening... (ENTER to send or ESC to cancel)"), - m.activeTab.editor.SetRecording(true), - m.waitForTranscript(), - ) -} - -func (m *appModel) handleStopSpeak() (tea.Model, tea.Cmd) { - if !m.transcriber.IsRunning() { - return m, nil - } - - m.transcriber.Stop() - m.closeTranscriptCh() - - return m, tea.Batch(m.activeTab.editor.SetRecording(false), notification.SuccessCmd("Stopped listening")) -} - -// waitForTranscript returns a command that blocks until the next transcript -// delta arrives and delivers it as a SpeakTranscriptMsg. -func (m *appModel) waitForTranscript() tea.Cmd { - ch := m.transcriptCh - return func() tea.Msg { - delta, ok := <-ch - if !ok { - return nil - } - return messages.SpeakTranscriptMsg{Delta: delta} - } -} - -// closeTranscriptCh closes the transcript channel and sets it to nil, -// unblocking any goroutines waiting in waitForTranscript. -func (m *appModel) closeTranscriptCh() { - if m.transcriptCh != nil { - close(m.transcriptCh) - m.transcriptCh = nil - } -} - func (m *appModel) handleElicitationResponse(action tools.ElicitationAction, content map[string]any, elicitationID string) (tea.Model, tea.Cmd) { if err := m.application.ResumeElicitation(m.ctx(), action, content, elicitationID); err != nil { slog.Error("Failed to resume elicitation", "action", action, "error", err) diff --git a/pkg/tui/messages/input.go b/pkg/tui/messages/input.go index 6b4326be74..1641e2e1b4 100644 --- a/pkg/tui/messages/input.go +++ b/pkg/tui/messages/input.go @@ -14,9 +14,6 @@ type ( // StopSpeakMsg stops speech-to-text transcription. StopSpeakMsg struct{} - // SpeakTranscriptMsg contains transcription delta from speech-to-text. - SpeakTranscriptMsg struct{ Delta string } - // StartShellMsg starts an interactive shell. StartShellMsg struct{} diff --git a/pkg/tui/speech.go b/pkg/tui/speech.go new file mode 100644 index 0000000000..9cf110f3ad --- /dev/null +++ b/pkg/tui/speech.go @@ -0,0 +1,211 @@ +package tui + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + + tea "charm.land/bubbletea/v2" + + "github.com/docker/docker-agent/pkg/tui/components/editor" + "github.com/docker/docker-agent/pkg/tui/components/notification" + "github.com/docker/docker-agent/pkg/tui/messages" +) + +type speechRun struct { + cancel context.CancelFunc + finish chan struct{} + done chan struct{} + wake chan struct{} + tab *tabModel + editor editor.Editor + err error // Read only after done closes. + + mu sync.Mutex + pending strings.Builder + ready bool + + // Only the UI loop requests completion and decides whether to send the draft. + stopping bool + send bool +} + +type speechEvent struct{ run *speechRun } + +func (r *speechRun) notify() { + select { + case r.wake <- struct{}{}: + default: + } +} + +func (m *appModel) isTranscribing() bool { + m.speechMu.Lock() + defer m.speechMu.Unlock() + return m.speech != nil +} + +func (m *appModel) handleStartSpeak() (tea.Model, tea.Cmd) { + m.speechMu.Lock() + defer m.speechMu.Unlock() + if m.speech != nil || m.speechCtx == nil || m.speechCtx.Err() != nil { + return m, nil + } + ctx, cancel := context.WithCancel(m.speechCtx) + run := &speechRun{ + cancel: cancel, + finish: make(chan struct{}), + done: make(chan struct{}), + wake: make(chan struct{}, 1), + tab: m.activeTab, + editor: m.activeTab.editor, + } + m.speech = run + go func() { + defer close(run.done) + defer cancel() + run.err = m.transcriber.Run(ctx, run.finish, func(_ context.Context, delta string) { + run.mu.Lock() + run.pending.WriteString(delta) + run.mu.Unlock() + run.notify() + }, func(context.Context) { + run.mu.Lock() + run.ready = true + run.mu.Unlock() + run.notify() + }) + }() + return m, waitForSpeech(run) +} + +func waitForSpeech(run *speechRun) tea.Cmd { + return func() tea.Msg { + select { + case <-run.wake: + case <-run.done: + } + return speechEvent{run: run} + } +} + +func (m *appModel) handleSpeechEvent(event speechEvent) (tea.Model, tea.Cmd) { + m.speechMu.Lock() + defer m.speechMu.Unlock() + if m.speech != event.run { + return m, nil + } + run := event.run + var done bool + select { + case <-run.done: + done = true + default: + } + // Notifications carry no text: even a notification in flight cannot lose a delta. + run.mu.Lock() + text, ready := run.pending.String(), run.ready + run.pending.Reset() + run.ready = false + run.mu.Unlock() + if text != "" { + run.editor.InsertText(text) + } + + if done { + m.speech = nil + var result tea.Cmd + switch { + case run.err != nil: + result = notification.ErrorCmd(fmt.Sprintf("Transcription failed: %v", run.err)) + case run.send && m.activeTab == run.tab && m.activeTab.editor == run.editor: + result = m.sendSpeechDraft(run) + case run.stopping: + result = notification.SuccessCmd("Stopped listening") + } + return m, tea.Batch(run.editor.SetRecording(false), result) + } + + var started tea.Cmd + if ready && !run.stopping { + started = tea.Batch(run.editor.SetRecording(true), notification.InfoCmd("🎤 Listening... (ENTER to send or ESC to cancel)")) + } + return m, tea.Batch(started, waitForSpeech(run)) +} + +// stopSpeech discards a run whose editor or application is being torn down. +func (m *appModel) stopSpeech() *speechRun { + return m.stopSpeechForTab(nil) +} + +func (m *appModel) stopSpeechForTab(tab *tabModel) *speechRun { + m.speechMu.Lock() + defer m.speechMu.Unlock() + run := m.speech + if run != nil && tab != nil && run.tab != tab { + return nil + } + if run != nil { + run.cancel() + <-run.done + m.speech = nil + if run.err != nil { + slog.Warn("Failed to stop transcription", "error", run.err) + } + } + return run +} + +func (m *appModel) handleStopSpeak() (tea.Model, tea.Cmd) { + m.speechMu.Lock() + defer m.speechMu.Unlock() + if run := m.speech; run != nil { + run.stopping, run.send = true, false + run.cancel() + } + // The existing subscription applies pending text and reports the final outcome. + return m, nil +} + +func (m *appModel) handleFinishSpeak() (tea.Model, tea.Cmd) { + m.speechMu.Lock() + defer m.speechMu.Unlock() + if run := m.speech; run != nil && !run.stopping { + run.stopping = true + run.send = m.activeTab == run.tab + close(run.finish) + return m, notification.InfoCmd("Finishing transcription… (ESC to cancel)") + } + return m, nil +} + +func (m *appModel) sendSpeechDraft(run *speechRun) tea.Cmd { + cmd := run.editor.SendContent() + if cmd == nil { + return nil + } + var tabID string + for id, tab := range m.tabs { + if tab == run.tab { + tabID = id + break + } + } + return func() tea.Msg { + return messages.RoutedMsg{ + SessionID: tabID, + Inner: cmd(), + Valid: func() bool { return m.tabs[tabID] == run.tab && run.tab.editor == run.editor }, + } + } +} + +func (m *appModel) cancelSpeechSend(tab *tabModel) { + m.speechMu.Lock() + defer m.speechMu.Unlock() + if run := m.speech; run != nil && run.tab == tab { + run.send = false + } +} diff --git a/pkg/tui/tui.go b/pkg/tui/tui.go index 85981317a6..a57d490227 100644 --- a/pkg/tui/tui.go +++ b/pkg/tui/tui.go @@ -166,7 +166,10 @@ type appModel struct { // Speech-to-text transcriber Transcriber - transcriptCh chan string // bridges transcriber goroutine → Bubble Tea event loop + speechMu sync.Mutex + speechCtx context.Context //nolint:containedctx // Owns transcription across UI and external shutdown. + cancelSpeech context.CancelFunc + speech *speechRun // Working state indicator (resize handle spinner) workingSpinner spinner.Spinner @@ -310,9 +313,7 @@ type themeFileWatcher interface { // can inject a fake implementation via WithTranscriber and so that the TUI // does not depend on a concrete audio backend. type Transcriber interface { - Start(ctx context.Context, handler transcribe.TranscriptHandler) error - Stop() - IsRunning() bool + Run(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, ready func(context.Context)) error IsSupported() bool } @@ -481,6 +482,7 @@ func WithToolRenderers(renderers map[string]tool.Builder) Option { func New(ctx context.Context, spawner SessionSpawner, initialApp *app.App, initialWorkingDir string, cleanup func(), opts ...Option) tea.Model { tuiCtx := func() context.Context { return context.WithoutCancel(ctx) } themeCtx, cancelThemes := context.WithCancel(ctx) + speechCtx, cancelSpeech := context.WithCancel(ctx) ar := animation.NewRuntime() @@ -514,6 +516,8 @@ func New(ctx context.Context, spawner SessionSpawner, initialApp *app.App, initi shutdownDone: ctx.Done(), themeCtx: themeCtx, cancelThemes: cancelThemes, + speechCtx: speechCtx, + cancelSpeech: cancelSpeech, themeChanges: make(chan messages.ThemeFileChangedMsg), toolRenderers: tooldefaults.NewRegistry(), buildCommandCategories: func(ctx context.Context, _ tea.Model) []commands.Category { @@ -737,6 +741,9 @@ func (m *appModel) editorOpts() []editor.Option { func (m *appModel) initSessionComponents(tabID string, a *app.App, sess *session.Session) { m.cancelSidebarPlanEdit() tab := m.ensureTab(tabID) + if run := m.stopSpeechForTab(tab); run != nil { + run.editor.SetRecording(false) + } if tab.chatPage != nil { m.retireAttention(tab) chat.Cleanup(tab.chatPage) @@ -1555,10 +1562,8 @@ func (m *appModel) update(msg tea.Msg) (tea.Model, tea.Cmd) { case messages.StopSpeakMsg: return m.handleStopSpeak() - case messages.SpeakTranscriptMsg: - m.activeTab.editor.InsertText(msg.Delta) - cmd := m.waitForTranscript() - return m, cmd + case speechEvent: + return m.handleSpeechEvent(msg) // --- MCP prompts --- @@ -2027,6 +2032,7 @@ func (m *appModel) handleSwitchTab(sessionID string) (tea.Model, tea.Cmd) { } m.cancelSidebarPlanEdit() if outgoing != m.tabs[sessionID] { + m.cancelSpeechSend(outgoing) m.parkAttention(outgoing) } @@ -2155,6 +2161,7 @@ func (m *appModel) handleCloseTab(sessionID string) (tea.Model, tea.Cmd) { // Clean up per-session state if tab := m.tabs[sessionID]; tab != nil { + m.stopSpeechForTab(tab) if tab.chatPage != nil { chat.Cleanup(tab.chatPage) } @@ -2389,12 +2396,10 @@ func (m *appModel) handleKeyPress(msg tea.KeyPressMsg) (tea.Model, tea.Cmd) { return m, nil } // Check if we should stop transcription on Enter or Escape - if m.transcriber.IsRunning() { + if m.isTranscribing() { switch { case key.Matches(msg, transcriptionEnterKey): - model, cmd := m.handleStopSpeak() - sendCmd := m.activeTab.editor.SendContent() - return model, tea.Batch(cmd, sendCmd) + return m.handleFinishSpeak() case key.Matches(msg, transcriptionEscapeKey): return m.handleStopSpeak() @@ -3217,6 +3222,10 @@ func (m *appModel) shutdownTimeoutOrDefault() time.Duration { // exit) and the context watcher (external cancellation). func (m *appModel) cleanupManagedResources() { m.cleanupOnce.Do(func() { + if m.cancelSpeech != nil { + m.cancelSpeech() + } + m.stopSpeech() if m.cancelThemes != nil { m.cancelThemes() } @@ -3253,8 +3262,9 @@ func (m *appModel) Shutdown() { // wedged cleanup or arm parallel safety nets. func (m *appModel) cleanupAll() { m.cleanupAllOnce.Do(func() { - m.transcriber.Stop() - m.closeTranscriptCh() + if m.cancelSpeech != nil { + m.cancelSpeech() + } for _, tab := range m.tabs { if tab.editor != nil { tab.editor.Cleanup() diff --git a/pkg/tui/tui_speak_test.go b/pkg/tui/tui_speak_test.go index 448f3c4108..b32bc993dc 100644 --- a/pkg/tui/tui_speak_test.go +++ b/pkg/tui/tui_speak_test.go @@ -3,139 +3,426 @@ package tui import ( "context" "errors" + "io" + "strings" + "sync/atomic" "testing" + "testing/synctest" + "time" + + tea "charm.land/bubbletea/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/docker/docker-agent/pkg/audio/transcribe" "github.com/docker/docker-agent/pkg/tui/components/notification" - "github.com/docker/docker-agent/pkg/tui/dialog" + "github.com/docker/docker-agent/pkg/tui/messages" ) -// fakeTranscriber is a controllable implementation of the Transcriber -// interface used to exercise the speech-to-text handlers without touching -// audio hardware or network. It is intentionally lock-free: the TUI runs the -// transcriber on the single Bubble Tea event-loop goroutine, so concurrent -// access from production code is not a concern, and tests interact with it -// strictly after each handler call returns. type fakeTranscriber struct { - supported bool - startErr error - running bool - startCalls int - stopCalls int + supported bool + err error + calls atomic.Int32 + cleaned atomic.Bool + handler func(context.Context, <-chan struct{}, transcribe.TranscriptHandler, func(context.Context)) error } func (f *fakeTranscriber) IsSupported() bool { return f.supported } -func (f *fakeTranscriber) IsRunning() bool { return f.running } - -func (f *fakeTranscriber) Start(_ context.Context, _ transcribe.TranscriptHandler) error { - f.startCalls++ - if f.startErr != nil { - return f.startErr +func (f *fakeTranscriber) Run(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, ready func(context.Context)) error { + f.calls.Add(1) + defer f.cleaned.Store(true) + if f.handler != nil { + return f.handler(ctx, finish, handler, ready) + } + if f.err != nil { + return f.err + } + ready(ctx) + select { + case <-ctx.Done(): + case <-finish: } - f.running = true return nil } -func (f *fakeTranscriber) Stop() { - f.stopCalls++ - f.running = false +func newSpeakTestModel(t *testing.T, ft *fakeTranscriber) *appModel { + t.Helper() + m, _ := newTestModel(t) + m.transcriber = ft + m.speechCtx, m.cancelSpeech = context.WithCancel(t.Context()) + t.Cleanup(m.cleanupManagedResources) + return m +} + +func nextSpeechEvent(t *testing.T, cmd tea.Cmd) speechEvent { + t.Helper() + require.NotNil(t, cmd) + event, ok := cmd().(speechEvent) + require.True(t, ok) + return event } -// newSpeakTestModel builds an appModel wired with a fakeTranscriber so that the -// speech-to-text handlers can be tested in isolation. It leverages the same -// minimal scaffolding used by the exit tests. -func newSpeakTestModel(tb testing.TB, ft *fakeTranscriber) *appModel { - tb.Helper() - page := &mockChatPage{} - ed := &mockEditor{} - tab := &tabModel{chatPage: page, editor: ed} +func waitSpeechDone(t *testing.T, run *speechRun) { + t.Helper() + select { + case <-run.done: + case <-time.After(5 * time.Second): + t.Fatal("transcription did not finish") + } +} - return &appModel{ - ctx: tb.Context, - tabs: map[string]*tabModel{"test": tab}, - activeTab: tab, - transcriber: ft, - dialogMgr: dialog.New(), +func requireSpeechError(t *testing.T, cmd tea.Cmd, text string) { + t.Helper() + msgs := collectMsgs(cmd) + for _, msg := range msgs { + if show, ok := msg.(notification.ShowMsg); ok { + require.Equal(t, notification.TypeError, show.Type) + require.Contains(t, show.Text, text) + return + } } + t.Fatal("missing error notification") } func TestHandleStartSpeak_NoOpIfAlreadyRunning(t *testing.T) { t.Parallel() - - ft := &fakeTranscriber{running: true} + ft := &fakeTranscriber{} m := newSpeakTestModel(t, ft) + _, cmd := m.handleStartSpeak() + nextSpeechEvent(t, cmd) + _, cmd = m.handleStartSpeak() + assert.Nil(t, cmd) + assert.EqualValues(t, 1, ft.calls.Load()) + m.stopSpeech() + assert.True(t, ft.cleaned.Load()) +} +func TestSpeech_StopPreservesStartupError(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{err: errors.New("missing API key")}) _, cmd := m.handleStartSpeak() - if cmd != nil { - t.Errorf("expected nil cmd when already running, got %T", cmd) - } - if ft.startCalls != 0 { - t.Errorf("Start should not be called when already running; got %d calls", ft.startCalls) - } + run := m.speech + waitSpeechDone(t, run) + terminal := nextSpeechEvent(t, cmd) + m.handleStopSpeak() + _, cmd = m.handleSpeechEvent(terminal) + assert.False(t, m.isTranscribing()) + requireSpeechError(t, cmd, "missing API key") +} + +func TestHandleStopSpeak_NoOpWhenNotRunning(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{}) + _, cmd := m.handleStopSpeak() + assert.Nil(t, cmd) } -func TestHandleStartSpeak_ReturnsErrorNotificationOnStartFailure(t *testing.T) { +func TestSpeech_StopRetainsInflightText(t *testing.T) { t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + handler(ctx, " final words") + <-ctx.Done() + return nil + }}) + ed := &speechEditor{text: "first words"} + m.activeTab.editor = ed + _, cmd := m.handleStartSpeak() + run := m.speech + inflight := nextSpeechEvent(t, cmd) + m.handleStopSpeak() + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(inflight) + assert.Equal(t, "first words final words", ed.text) + assert.False(t, m.isTranscribing()) + _, cmd = m.handleSpeechEvent(inflight) + assert.Nil(t, cmd) + assert.Equal(t, "first words final words", ed.text) +} + +func TestSpeech_IgnoresDiscardedRun(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{}) + _, _ = m.handleStartSpeak() + previous := m.stopSpeech() + _, _ = m.handleStartSpeak() + _, cmd := m.handleSpeechEvent(speechEvent{run: previous}) + assert.Nil(t, cmd) + assert.True(t, m.isTranscribing()) +} - ft := &fakeTranscriber{startErr: errors.New("boom")} +func TestSpeech_ShutdownJoinsCapture(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + ft := &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, _ transcribe.TranscriptHandler, ready func(context.Context)) error { + ready(ctx) + <-ctx.Done() + <-release + return nil + }} + m := newSpeakTestModel(t, ft) + _, cmd := m.handleStartSpeak() + nextSpeechEvent(t, cmd) + done := make(chan struct{}) + go func() { + defer close(done) + m.cleanupManagedResources() + }() + synctest.Wait() + assert.False(t, ft.cleaned.Load()) + select { + case <-done: + t.Fatal("shutdown returned before capture cleanup") + default: + } + close(release) + <-done + assert.True(t, ft.cleaned.Load()) + _, cmd = m.handleStartSpeak() + assert.Nil(t, cmd) + }) +} + +func TestSpeech_StopDoesNotBlockUI(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, _ transcribe.TranscriptHandler, ready func(context.Context)) error { + ready(ctx) + <-ctx.Done() + <-release + return nil + }}) + _, cmd := m.handleStartSpeak() + run := m.speech + nextSpeechEvent(t, cmd) + m.handleStopSpeak() + synctest.Wait() + assert.True(t, m.isTranscribing(), "keep the run until cleanup completes") + _, cmd = m.handleStartSpeak() + assert.Nil(t, cmd, "do not overlap capture during cleanup") + close(release) + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(speechEvent{run: run}) + assert.False(t, m.isTranscribing()) + }) +} + +func TestSpeech_ParentCancellation(t *testing.T) { + t.Parallel() + ft := &fakeTranscriber{} m := newSpeakTestModel(t, ft) + _, _ = m.handleStartSpeak() + run := m.speech + m.cancelSpeech() + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(speechEvent{run: run}) + assert.False(t, m.isTranscribing()) + assert.True(t, ft.cleaned.Load()) +} + +type speechEditor struct { + mockEditor + text string + recording bool + sent []string +} + +func (e *speechEditor) InsertText(text string) { e.text += text } +func (e *speechEditor) SetRecording(recording bool) tea.Cmd { + e.recording = recording + return nil +} + +func (e *speechEditor) SendContent() tea.Cmd { + e.sent = append(e.sent, e.text) + return nil +} + +func TestSpeech_FinishSendsAllTextOnce(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + handler(ctx, "first") + <-finish + handler(ctx, " last") + return nil + }}) + ed := &speechEditor{} + m.activeTab.editor = ed _, cmd := m.handleStartSpeak() - if cmd == nil { - t.Fatalf("expected an error notification cmd, got nil") - } - if ft.startCalls != 1 { - t.Errorf("Start should be called exactly once; got %d", ft.startCalls) - } - if m.transcriptCh != nil { - t.Error("transcriptCh should be cleared after a failed Start") - } + run := m.speech + inflight := nextSpeechEvent(t, cmd) + m.handleFinishSpeak() + m.handleFinishSpeak() + assert.Empty(t, ed.sent) + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(inflight) + assert.Equal(t, []string{"first last"}, ed.sent) + _, _ = m.handleSpeechEvent(inflight) + assert.Len(t, ed.sent, 1) +} - // The returned cmd should produce an error notification.ShowMsg. - msg := cmd() - show, ok := msg.(notification.ShowMsg) - if !ok { - t.Fatalf("expected notification.ShowMsg, got %T (%v)", msg, msg) - } - if show.Type != notification.TypeError { - t.Errorf("expected notification of TypeError, got %v", show.Type) - } - if show.Text == "" { - t.Error("error notification should carry a non-empty Text") - } +func TestSpeech_FinishFailureKeepsDraft(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + handler(ctx, "keep this") + <-finish + return errors.New("final transcription timed out") + }}) + ed := &speechEditor{} + m.activeTab.editor = ed + _, _ = m.handleStartSpeak() + run := m.speech + m.handleFinishSpeak() + waitSpeechDone(t, run) + _, cmd := m.handleSpeechEvent(speechEvent{run: run}) + requireSpeechError(t, cmd, "timed out") + assert.Equal(t, "keep this", ed.text) + assert.Empty(t, ed.sent) } -func TestHandleStopSpeak_NoOpWhenNotRunning(t *testing.T) { +func TestSpeech_SwitchTabWhileFinishingKeepsOriginDraft(t *testing.T) { t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + <-finish + handler(ctx, "origin") + return nil + }}) + origin := &speechEditor{} + m.activeTab.editor = origin + _, _ = m.handleStartSpeak() + run := m.speech + m.handleFinishSpeak() + other := &speechEditor{} + m.activeTab = &tabModel{editor: other} + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(speechEvent{run: run}) + assert.Equal(t, "origin", origin.text) + assert.Empty(t, origin.sent) + assert.Empty(t, other.text) + assert.Empty(t, other.sent) +} - ft := &fakeTranscriber{running: false} - m := newSpeakTestModel(t, ft) +func TestSpeech_EscapeCancelsPendingSend(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + <-finish + handler(ctx, "draft") + <-ctx.Done() + return nil + }}) + ed := &speechEditor{} + m.activeTab.editor = ed + _, _ = m.handleStartSpeak() + run := m.speech + m.handleFinishSpeak() + m.handleStopSpeak() + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(speechEvent{run: run}) + assert.Equal(t, "draft", ed.text) + assert.Empty(t, ed.sent) +} - _, cmd := m.handleStopSpeak() - if cmd != nil { - t.Errorf("expected nil cmd when not running, got %T", cmd) - } - if ft.stopCalls != 0 { - t.Errorf("Stop should not be called when not running; got %d", ft.stopCalls) +func TestSpeech_ExitSafetyNetCoversBlockedCleanup(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + ft := &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, _ transcribe.TranscriptHandler, ready func(context.Context)) error { + ready(ctx) + <-ctx.Done() + <-release + return nil + }} + m := newSpeakTestModel(t, ft) + t.Cleanup(func() { close(release) }) + _, cmd := m.handleStartSpeak() + nextSpeechEvent(t, cmd) + m.program = tea.NewProgram(m, tea.WithInput(nil), tea.WithOutput(io.Discard)) + defer m.program.Kill() + exited := make(chan struct{}, 1) + m.exitFunc = func(int) { exited <- struct{}{} } + m.shutdownTimeout = 10 * time.Millisecond + m.cleanupAll() + select { + case <-exited: + case <-time.After(5 * time.Second): + t.Fatal("speech cleanup blocked the exit safety net") } } -func TestHandleStopSpeak_StopsAndNotifies(t *testing.T) { +func TestSpeech_StopPreservesCleanupError(t *testing.T) { t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, _ transcribe.TranscriptHandler, ready func(context.Context)) error { + ready(ctx) + <-ctx.Done() + return errors.New("dispose failed") + }}) + _, cmd := m.handleStartSpeak() + run := m.speech + nextSpeechEvent(t, cmd) + m.handleStopSpeak() + waitSpeechDone(t, run) + _, cmd = m.handleSpeechEvent(speechEvent{run: run}) + requireSpeechError(t, cmd, "dispose failed") +} - ft := &fakeTranscriber{running: true} - m := newSpeakTestModel(t, ft) - // Pretend a previous start opened a channel. - m.transcriptCh = make(chan string, 1) +type routedSpeechEditor struct{ speechEditor } - _, cmd := m.handleStopSpeak() - if cmd == nil { - t.Fatalf("expected a batch cmd, got nil") - } - if ft.stopCalls != 1 { - t.Errorf("Stop should be called exactly once; got %d", ft.stopCalls) - } - if m.transcriptCh != nil { - t.Error("transcriptCh should be cleared after Stop") - } +func (e *routedSpeechEditor) SendContent() tea.Cmd { + content := e.text + return func() tea.Msg { return messages.SendMsg{Content: content} } +} + +func TestSpeech_SendStaysScopedUntilConsumption(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{}) + ed := &routedSpeechEditor{} + ed.text = "dictation" + m.activeTab.editor = ed + run := &speechRun{tab: m.activeTab, editor: ed} + cmd := m.sendSpeechDraft(run) + origin := m.activeTab + m.activeTab = &tabModel{editor: &speechEditor{}} + routed, ok := cmd().(messages.RoutedMsg) + require.True(t, ok) + assert.Equal(t, "test", routed.SessionID) + assert.Equal(t, messages.SendMsg{Content: "dictation"}, routed.Inner) + assert.True(t, routed.Valid()) + origin.editor = &speechEditor{} + assert.False(t, routed.Valid(), "a replacement editor must invalidate the queued send") +} + +func TestSpeech_CoalescedNotificationsLoseNoText(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, _ <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + for range 100 { + handler(ctx, "x") + } + return nil + }}) + ed := &speechEditor{} + m.activeTab.editor = ed + _, cmd := m.handleStartSpeak() + run := m.speech + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(nextSpeechEvent(t, cmd)) + assert.Equal(t, strings.Repeat("x", 100), ed.text) + assert.False(t, m.isTranscribing()) +} + +func TestSpeech_LeavingAndReturningRevokesSend(t *testing.T) { + t.Parallel() + m := newSpeakTestModel(t, &fakeTranscriber{handler: func(ctx context.Context, finish <-chan struct{}, handler transcribe.TranscriptHandler, _ func(context.Context)) error { + <-finish + handler(ctx, "keep draft") + return nil + }}) + ed := &speechEditor{} + m.activeTab.editor = ed + _, _ = m.handleStartSpeak() + run := m.speech + m.handleFinishSpeak() + m.cancelSpeechSend(m.activeTab) + waitSpeechDone(t, run) + _, _ = m.handleSpeechEvent(speechEvent{run: run}) + assert.Equal(t, "keep draft", ed.text) + assert.Empty(t, ed.sent) }