diff --git a/internal/dictation/transcriber_deepgram.go b/internal/dictation/transcriber_deepgram.go index d4a69ce7f..fd3a6d00f 100644 --- a/internal/dictation/transcriber_deepgram.go +++ b/internal/dictation/transcriber_deepgram.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" + "github.com/Gitlawb/zero/internal/providers/providerio" "github.com/coder/websocket" ) @@ -60,7 +61,15 @@ func (d *deepgramTranscriber) StreamTranscribe(ctx context.Context, chunks <-cha HTTPHeader: http.Header{"Authorization": {"Token " + d.cfg.APIKey}}, }) if err != nil { - return "", fmt.Errorf("connecting to Deepgram: %w", err) + // Preserve the context.Canceled sentinel (it carries no key) so an + // Esc-abort during dial still matches the UI's errors.Is check. Key + // on ctx.Err() itself rather than unwrapping err: a cancellation can + // surface as a plain transport error (e.g. "closed network + // connection") rather than context.Canceled directly. + if ctx.Err() != nil { + return "", ctx.Err() + } + return "", fmt.Errorf("connecting to Deepgram: %s", providerio.Redact(err.Error(), d.cfg.APIKey)) } defer conn.CloseNow() @@ -68,15 +77,26 @@ func (d *deepgramTranscriber) StreamTranscribe(ctx context.Context, chunks <-cha // so no conversion (unlike sherpa-onnx, which wants float32). writeErrCh := make(chan error, 1) go func() { - for chunk := range chunks { - if err := conn.Write(ctx, websocket.MessageBinary, chunk); err != nil { - writeErrCh <- err + // Select on ctx.Done() while waiting for the next chunk so a cancelled + // session does not leave this goroutine blocked on an open idle channel. + for { + select { + case <-ctx.Done(): + writeErrCh <- ctx.Err() return + case chunk, ok := <-chunks: + if !ok { + // CloseStream flushes any buffered audio and returns final + // results before Deepgram closes the socket. + writeErrCh <- conn.Write(ctx, websocket.MessageText, []byte(`{"type":"CloseStream"}`)) + return + } + if err := conn.Write(ctx, websocket.MessageBinary, chunk); err != nil { + writeErrCh <- err + return + } } } - // CloseStream flushes any buffered audio and returns final results before - // Deepgram closes the socket. - writeErrCh <- conn.Write(ctx, websocket.MessageText, []byte(`{"type":"CloseStream"}`)) }() // Deepgram results are per-utterance segments, not cumulative: accumulate the @@ -100,7 +120,18 @@ func (d *deepgramTranscriber) StreamTranscribe(ctx context.Context, chunks <-cha } default: } - return compose(), fmt.Errorf("Deepgram stream error: %w", err) //nolint:staticcheck // Preserve established user-facing error text. + // A user abort cancels the streaming context; the UI matches it + // with errors.Is(err, context.Canceled), so return the sentinel + // itself (it carries no key) instead of a flat redacted string. + // Key on ctx.Err() rather than unwrapping err: the writeErrCh + // swap above can replace err with a plain transport error (e.g. + // "closed network connection") that doesn't itself unwrap to + // context.Canceled even though the cancellation is what caused it. + if ctx.Err() != nil { + return compose(), ctx.Err() + } + //nolint:staticcheck // Preserve established user-facing error text. + return compose(), fmt.Errorf("Deepgram stream error: %s", providerio.Redact(err.Error(), d.cfg.APIKey)) } if typ != websocket.MessageText { continue diff --git a/internal/dictation/transcriber_deepgram_test.go b/internal/dictation/transcriber_deepgram_test.go new file mode 100644 index 000000000..4ee8880dc --- /dev/null +++ b/internal/dictation/transcriber_deepgram_test.go @@ -0,0 +1,278 @@ +package dictation + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/coder/websocket" +) + +// Non-cancellation dial failure: handshake returns 101 with an Upgrade header +// equal to the API key so the dial error embeds the key and redaction must strip it. +func TestDeepgramDialFailureRedactsKey(t *testing.T) { + const key = "sk-test-dial-key-abcdef" + baseURL := dialFailServerWithUpgrade(t, key) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: key, BaseURL: baseURL}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, nil) + if ferr == nil { + t.Fatal("expected dial failure") + } + if errors.Is(ferr, context.Canceled) { + t.Fatalf("dial failure should not be context.Canceled: %v", ferr) + } + if strings.Contains(ferr.Error(), key) { + t.Errorf("API key leaked from dial failure: %v", ferr) + } + if !strings.Contains(ferr.Error(), "connecting to Deepgram") { + t.Errorf("expected dial-path error prefix, got: %v", ferr) + } +} + +// dialFailServerWithUpgrade returns a ws URL whose handshake responds 101 with +// Upgrade set to upgradeValue (not "websocket"), so websocket.Dial fails with +// that value in the error string. Hijacks and closes immediately to avoid the +// library's 3s body-drain timeout on failed dials. +func dialFailServerWithUpgrade(t *testing.T, upgradeValue string) string { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "hijack not supported", http.StatusInternalServerError) + return + } + conn, bufrw, err := hj.Hijack() + if err != nil { + return + } + defer conn.Close() + _, _ = bufrw.WriteString("HTTP/1.1 101 Switching Protocols\r\n") + _, _ = bufrw.WriteString("Connection: Upgrade\r\n") + _, _ = bufrw.WriteString("Upgrade: " + upgradeValue + "\r\n\r\n") + _ = bufrw.Flush() + })) + t.Cleanup(srv.Close) + return "ws" + strings.TrimPrefix(srv.URL, "http") +} + +func TestDeepgramStreamTranscribeErrorRedaction(t *testing.T) { + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + // Drain the client's audio frames, then wait for CloseStream so the + // client has flushed its writes before we reject the connection. Closing + // immediately (before CloseStream) risks the client failing on a write + // error and never observing this API-key-bearing close reason. + for { + typ, data, err := c.Read(ctx) + if err != nil { + return + } + if typ == websocket.MessageText && strings.Contains(string(data), "CloseStream") { + break + } + } + c.Close(websocket.StatusPolicyViolation, "invalid key sk-test-key") + }) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte, 1) + chunks <- make([]byte, 320) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, func(string, bool) {}) + if ferr == nil { + t.Fatal("expected error") + } + if strings.Contains(ferr.Error(), "sk-test-key") { + t.Errorf("API key leaked: %v", ferr) + } +} + +func TestDeepgramStreamTranscribeCancelKeepsSentinel(t *testing.T) { + firstFrame := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + // Hold the connection open, never answering, so the client blocks in + // Read until its context is cancelled (the Esc-abort path). + var once sync.Once + for { + if _, _, err := c.Read(ctx); err != nil { + return + } + once.Do(func() { close(firstFrame) }) + } + }) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + defer close(chunks) + chunks <- make([]byte, 320) + // The channel stays open: the session is live when the user aborts. + errCh := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + errCh <- ferr + }() + + select { + case <-firstFrame: + cancel() + ferr := <-errCh + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("cancelled stream error lost the context.Canceled sentinel: %v", ferr) + } + case ferr := <-errCh: + t.Fatalf("StreamTranscribe failed early instead of blocking: %v", ferr) + } +} + +// Esc can cancel while the WebSocket dial is still pending. The dial error is +// redacted with %s, so we must return ctx.Err() rather than a flat string. +func TestDeepgramStreamTranscribeDialCancelKeepsSentinel(t *testing.T) { + tr, err := NewDeepgramTranscriber(DeepgramConfig{ + APIKey: "sk-test-key", + BaseURL: "ws://127.0.0.1:1", // never reached; ctx is already cancelled + }) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + chunks := make(chan []byte) + defer close(chunks) + + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("dial-cancel lost the context.Canceled sentinel: %v", ferr) + } +} + +// Esc right after accept can race the first write or the first Read; both +// setup/write redaction sites must still yield context.Canceled. +func TestDeepgramStreamTranscribeStartupCancelKeepsSentinel(t *testing.T) { + accepted := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + close(accepted) + for { + if _, _, err := c.Read(ctx); err != nil { + return + } + } + }) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + defer close(chunks) + // No frames yet: the writer blocks on chunks or the reader blocks on Read. + errCh := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + errCh <- ferr + }() + + select { + case <-accepted: + cancel() + ferr := <-errCh + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("startup-cancel lost the context.Canceled sentinel: %v", ferr) + } + case ferr := <-errCh: + t.Fatalf("StreamTranscribe failed before accept: %v", ferr) + } +} + +// TestDeepgramStreamErrorRedactsHeaderShapedKey exercises the same stream-error +// redaction path as TestDeepgramStreamTranscribeErrorRedaction, with a close +// reason that embeds the key next to an X-Api-Key-shaped header label. +func TestDeepgramStreamErrorRedactsHeaderShapedKey(t *testing.T) { + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + for { + typ, data, err := c.Read(ctx) + if err != nil { + return + } + if typ == websocket.MessageText && strings.Contains(string(data), "CloseStream") { + break + } + } + c.Close(websocket.StatusPolicyViolation, "X-Api-Key: sk-custom-secret-key-1234567890 failed") + }) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: "sk-custom-secret-key-1234567890", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte, 1) + chunks <- make([]byte, 320) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, func(string, bool) {}) + if ferr == nil { + t.Fatal("expected error") + } + if strings.Contains(ferr.Error(), "sk-custom-secret-key-1234567890") { + t.Errorf("Custom header API key leaked: %v", ferr) + } +} + +func TestDeepgramSinglePassRedaction(t *testing.T) { + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + for { + typ, data, err := c.Read(ctx) + if err != nil { + return + } + if typ == websocket.MessageText && strings.Contains(string(data), "CloseStream") { + break + } + } + c.Close(websocket.StatusPolicyViolation, "error with sk-key1234567890abcdef1234") + }) + + tr, err := NewDeepgramTranscriber(DeepgramConfig{APIKey: "sk-key1234567890abcdef1234", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte, 1) + chunks <- make([]byte, 320) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, func(string, bool) {}) + if ferr == nil { + t.Fatal("expected error") + } + got := ferr.Error() + if strings.Contains(got, "sk-key1234567890abcdef1234") { + t.Fatalf("API key leaked into stream error: %q", got) + } + // One key occurrence must produce exactly one redaction marker (no double wrap). + if strings.Count(got, "[REDACTED]") != 1 { + t.Fatalf("redaction markers in %q: want exactly 1 [REDACTED]", got) + } +} diff --git a/internal/dictation/transcriber_openai_realtime.go b/internal/dictation/transcriber_openai_realtime.go index 590cbc8b4..ae737d43c 100644 --- a/internal/dictation/transcriber_openai_realtime.go +++ b/internal/dictation/transcriber_openai_realtime.go @@ -8,6 +8,7 @@ import ( "net/http" "strings" + "github.com/Gitlawb/zero/internal/providers/providerio" "github.com/coder/websocket" ) @@ -22,6 +23,10 @@ type OpenAIRealtimeConfig struct { Model string // default "gpt-4o-transcribe" // BaseURL overrides the wss endpoint (tests point it at a fake server). BaseURL string + // writeErrInjector, when set, is consulted before each JSON websocket write. + // Used by same-package tests to force controllable write failures that embed + // secrets; production callers leave it nil. + writeErrInjector func() error } type openAIRealtimeTranscriber struct { @@ -59,7 +64,15 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks }, }) if err != nil { - return "", fmt.Errorf("connecting to OpenAI Realtime: %w", err) + // Preserve the context.Canceled sentinel (it carries no key) so an + // Esc-abort during dial still matches the UI's errors.Is check. Key + // on ctx.Err() itself rather than unwrapping err: a cancellation can + // surface as a plain transport error (e.g. "closed network + // connection") rather than context.Canceled directly. + if ctx.Err() != nil { + return "", ctx.Err() + } + return "", fmt.Errorf("connecting to OpenAI Realtime: %s", providerio.Redact(err.Error(), o.cfg.APIKey)) } defer conn.CloseNow() @@ -73,24 +86,40 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks }, }, } - if err := writeJSON(ctx, conn, sessionUpdate); err != nil { - return "", fmt.Errorf("configuring OpenAI Realtime session: %w", err) + if err := o.writeJSON(ctx, conn, sessionUpdate); err != nil { + // Same cancellation short-circuit as the dial above: an Esc-abort + // while the session update is in flight must stay context.Canceled. + if ctx.Err() != nil { + return "", ctx.Err() + } + return "", fmt.Errorf("configuring OpenAI Realtime session: %s", providerio.Redact(err.Error(), o.cfg.APIKey)) } writeErrCh := make(chan error, 1) go func() { - for chunk := range chunks { - appendMsg := map[string]any{ - "type": "input_audio_buffer.append", - "audio": base64.StdEncoding.EncodeToString(chunk), - } - if err := writeJSON(ctx, conn, appendMsg); err != nil { - writeErrCh <- err + // Select on ctx.Done() while waiting for the next chunk so a cancelled + // session does not leave this goroutine blocked on an open idle channel. + for { + select { + case <-ctx.Done(): + writeErrCh <- ctx.Err() return + case chunk, ok := <-chunks: + if !ok { + // Commit the buffered audio to force final transcription. + writeErrCh <- o.writeJSON(ctx, conn, map[string]any{"type": "input_audio_buffer.commit"}) + return + } + appendMsg := map[string]any{ + "type": "input_audio_buffer.append", + "audio": base64.StdEncoding.EncodeToString(chunk), + } + if err := o.writeJSON(ctx, conn, appendMsg); err != nil { + writeErrCh <- err + return + } } } - // Commit the buffered audio to force final transcription of the utterance. - writeErrCh <- writeJSON(ctx, conn, map[string]any{"type": "input_audio_buffer.commit"}) }() // OpenAI deltas are incremental (append), unlike Deepgram/sherpa's cumulative @@ -113,7 +142,17 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks } default: } - return compose(), fmt.Errorf("OpenAI Realtime stream error: %w", err) + // A user abort cancels the streaming context; the UI matches it + // with errors.Is(err, context.Canceled), so return the sentinel + // itself (it carries no key) instead of a flat redacted string. + // Key on ctx.Err() rather than unwrapping err: the writeErrCh + // swap above can replace err with a plain transport error (e.g. + // "closed network connection") that doesn't itself unwrap to + // context.Canceled even though the cancellation is what caused it. + if ctx.Err() != nil { + return compose(), ctx.Err() + } + return compose(), fmt.Errorf("OpenAI Realtime stream error: %s", providerio.Redact(err.Error(), o.cfg.APIKey)) } if typ != websocket.MessageText { continue @@ -125,6 +164,12 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks if onPartial != nil { onPartial(compose(), false) } + // onPartial may deliver the partial that makes the TUI cancel + // (Esc). Observe that before the next Read, which can already + // have a racing OpenAI error frame buffered. + if ctx.Err() != nil { + return compose(), ctx.Err() + } case realtimeCompleted: // A completed item replaces the in-progress delta buffer with the // server's authoritative transcript for that segment. @@ -138,6 +183,9 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks if onPartial != nil { onPartial(compose(), true) } + if ctx.Err() != nil { + return compose(), ctx.Err() + } // Once we've committed the buffer (user stopped) and the server has // returned a completed transcription, the utterance is done. if committed { @@ -147,7 +195,15 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks case realtimeCommitted: committed = true case realtimeError: - return compose(), fmt.Errorf("OpenAI Realtime error: %s", evt.text) + // Esc can race an incoming error event: conn.Read may already have + // the OpenAI error in hand by the time cancellation lands. Key on + // ctx.Err() first, same as every other return point in this loop, + // so a cancel that wins the race still surfaces as context.Canceled + // instead of a spurious redacted failure alongside it. + if ctx.Err() != nil { + return compose(), ctx.Err() + } + return compose(), fmt.Errorf("OpenAI Realtime error: %s", providerio.Redact(evt.text, o.cfg.APIKey)) } // The writer signals commit completion out of band; observe it so a // stop with no further audio still flips `committed`. @@ -155,6 +211,10 @@ func (o *openAIRealtimeTranscriber) StreamTranscribe(ctx context.Context, chunks case werr := <-writeErrCh: if werr == nil { committed = true + } else if ctx.Err() != nil { + return compose(), ctx.Err() + } else { + return compose(), fmt.Errorf("OpenAI Realtime stream error: %s", providerio.Redact(werr.Error(), o.cfg.APIKey)) } default: } @@ -201,6 +261,15 @@ func parseRealtimeEvent(data []byte) realtimeEvent { return realtimeEvent{kind: realtimeOther} } +func (o *openAIRealtimeTranscriber) writeJSON(ctx context.Context, conn *websocket.Conn, v any) error { + if o.cfg.writeErrInjector != nil { + if err := o.cfg.writeErrInjector(); err != nil { + return err + } + } + return writeJSON(ctx, conn, v) +} + func writeJSON(ctx context.Context, conn *websocket.Conn, v any) error { data, err := json.Marshal(v) if err != nil { diff --git a/internal/dictation/transcriber_openai_realtime_test.go b/internal/dictation/transcriber_openai_realtime_test.go new file mode 100644 index 000000000..4b7ba66f7 --- /dev/null +++ b/internal/dictation/transcriber_openai_realtime_test.go @@ -0,0 +1,379 @@ +package dictation + +import ( + "context" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +// Non-cancellation dial failure: handshake returns 101 with an Upgrade header +// equal to the API key so the dial error embeds the key and redaction must strip it. +func TestOpenAIRealtimeDialFailureRedactsKey(t *testing.T) { + const key = "sk-test-dial-key-abcdef" + baseURL := dialFailServerWithUpgrade(t, key) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: key, BaseURL: baseURL}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, nil) + if ferr == nil { + t.Fatal("expected dial failure") + } + if errors.Is(ferr, context.Canceled) { + t.Fatalf("dial failure should not be context.Canceled: %v", ferr) + } + if strings.Contains(ferr.Error(), key) { + t.Errorf("API key leaked from dial failure: %v", ferr) + } + if !strings.Contains(ferr.Error(), "connecting to OpenAI Realtime") { + t.Errorf("expected dial-path error prefix, got: %v", ferr) + } +} + +// Session-configuration write fails after dial: inject a write error that embeds +// the API key (peer close cannot reach this path reliably — the session write +// always lands in the kernel buffer first) and assert redaction. +func TestOpenAIRealtimeSessionConfigFailureRedactsKey(t *testing.T) { + const key = "sk-test-session-key-abcdef" + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + // Hold open; the client fails on the injected session write before reading. + <-ctx.Done() + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{ + APIKey: key, + BaseURL: url, + writeErrInjector: func() error { + return errors.New("invalid API key " + key) + }, + }) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, nil) + if ferr == nil { + t.Fatal("expected session-config failure") + } + if errors.Is(ferr, context.Canceled) { + t.Fatalf("session-config failure should not be context.Canceled: %v", ferr) + } + if strings.Contains(ferr.Error(), key) { + t.Errorf("API key leaked from session-config failure: %v", ferr) + } + if !strings.Contains(ferr.Error(), "configuring OpenAI Realtime session") { + t.Errorf("expected session-config error prefix, got: %v", ferr) + } +} + +// Asynchronous writer failure: session update succeeds, then the audio writer +// injects a key-bearing write error. The server keeps the connection open and +// emits a delta so the reader loop observes writeErrCh (not a peer close). +func TestOpenAIRealtimeWriterFailureRedactsKey(t *testing.T) { + const key = "sk-test-writer-key-abcdef" + sessionDone := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + if _, _, err := c.Read(ctx); err != nil { + return + } + close(sessionDone) + // Keep reading and emit a delta so the client processes an event and + // drains writeErrCh while the connection is still open. + _ = c.Write(ctx, websocket.MessageText, []byte( + `{"type":"conversation.item.input_audio_transcription.delta","delta":"hi"}`, + )) + for { + if _, _, err := c.Read(ctx); err != nil { + return + } + } + }) + + var writes atomic.Int32 + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{ + APIKey: key, + BaseURL: url, + writeErrInjector: func() error { + // First write is session.update; fail subsequent audio/commit writes. + if writes.Add(1) == 1 { + return nil + } + return errors.New("invalid API key " + key) + }, + }) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte, 1) + chunks <- make([]byte, 480) + // Leave open after the first frame so the writer hits the injected error + // on that append rather than a clean commit. + + done := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(context.Background(), chunks, func(string, bool) {}) + done <- ferr + }() + + var ferr error + select { + case ferr = <-done: + case <-time.After(5 * time.Second): + close(chunks) + t.Fatal("StreamTranscribe did not complete after writer failure") + } + close(chunks) + + if ferr == nil { + t.Fatal("expected writer failure") + } + if errors.Is(ferr, context.Canceled) { + t.Fatalf("writer failure should not be context.Canceled: %v", ferr) + } + if strings.Contains(ferr.Error(), key) { + t.Errorf("API key leaked from writer failure: %v", ferr) + } + if !strings.Contains(ferr.Error(), "OpenAI Realtime stream error") { + t.Errorf("expected stream/writer error prefix, got: %v", ferr) + } + select { + case <-sessionDone: + default: + // Session may race the injected audio write; prefix assertion above is enough. + } +} + +func TestOpenAIRealtimeStreamTranscribeErrorRedaction(t *testing.T) { + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + defer c.Close(websocket.StatusNormalClosure, "") + for { + typ, _, err := c.Read(ctx) + if err != nil { + return + } + if typ != websocket.MessageText { + continue + } + _ = c.Write(ctx, websocket.MessageText, []byte(`{"type":"error","error":{"message":"invalid API key sk-test-key"}}`)) + } + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + chunks := make(chan []byte, 1) + chunks <- make([]byte, 480) + close(chunks) + _, ferr := tr.StreamTranscribe(context.Background(), chunks, func(string, bool) {}) + if ferr == nil { + t.Fatal("expected error") + } + if strings.Contains(ferr.Error(), "sk-test-key") { + t.Errorf("API key leaked: %v", ferr) + } +} + +func TestOpenAIRealtimeStreamTranscribeCancelKeepsSentinel(t *testing.T) { + firstFrame := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + // Hold the connection open, never answering, so the client blocks in + // Read until its context is cancelled (the Esc-abort path). + var once sync.Once + for { + if _, _, err := c.Read(ctx); err != nil { + return + } + once.Do(func() { close(firstFrame) }) + } + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + defer close(chunks) + chunks <- make([]byte, 480) + // The channel stays open: the session is live when the user aborts. + errCh := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + errCh <- ferr + }() + + select { + case <-firstFrame: + cancel() + ferr := <-errCh + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("cancelled stream error lost the context.Canceled sentinel: %v", ferr) + } + case ferr := <-errCh: + t.Fatalf("StreamTranscribe failed early instead of blocking: %v", ferr) + } +} + +// Esc can cancel while the WebSocket dial is still pending. The dial error is +// redacted with %s, so we must return ctx.Err() rather than a flat string. +func TestOpenAIRealtimeStreamTranscribeDialCancelKeepsSentinel(t *testing.T) { + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{ + APIKey: "sk-test-key", + BaseURL: "ws://127.0.0.1:1", // never reached; ctx is already cancelled + }) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + chunks := make(chan []byte) + defer close(chunks) + + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("dial-cancel lost the context.Canceled sentinel: %v", ferr) + } +} + +// Esc right after accept can hit the session-update write or the first Read; +// both redaction sites must still yield context.Canceled. +func TestOpenAIRealtimeStreamTranscribeStartupCancelKeepsSentinel(t *testing.T) { + accepted := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + close(accepted) + // Do not read: leave the client in session-update write or first Read. + <-ctx.Done() + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + defer close(chunks) + errCh := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + errCh <- ferr + }() + + select { + case <-accepted: + cancel() + ferr := <-errCh + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("startup-cancel lost the context.Canceled sentinel: %v", ferr) + } + case ferr := <-errCh: + t.Fatalf("StreamTranscribe failed before accept: %v", ferr) + } +} + +// Cancel while the writer is blocked on more chunks. Whichever path observes +// the cancellation first (the blocked conn.Read or the writeErrCh drain) must +// return context.Canceled rather than a redacted flat string. +func TestOpenAIRealtimeStreamTranscribeWriteCancelKeepsSentinel(t *testing.T) { + sessionReceived := make(chan struct{}) + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + if _, _, err := c.Read(ctx); err != nil { + return + } + close(sessionReceived) + for { + if _, _, err := c.Read(ctx); err != nil { + return + } + } + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + chunks <- make([]byte, 480) + defer close(chunks) + + errCh := make(chan error, 1) + go func() { + _, ferr := tr.StreamTranscribe(ctx, chunks, nil) + errCh <- ferr + }() + + select { + case <-sessionReceived: + cancel() + ferr := <-errCh + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("write-path cancel lost the context.Canceled sentinel: %v", ferr) + } + case ferr := <-errCh: + t.Fatalf("StreamTranscribe failed early instead of blocking: %v", ferr) + } +} + +// Esc can race an incoming OpenAI error event. conn.Read may already have +// the error frame in hand by the time the cancellation lands. The server +// fires a delta immediately followed by the error, back to back. Cancel from +// inside the delta callback (same path the TUI uses when a partial triggers +// Esc handling) so the cancel is observed before the next event is processed. +// The returned error must still be context.Canceled, not the redacted OpenAI +// error. +func TestOpenAIRealtimeStreamTranscribeErrorRaceKeepsSentinel(t *testing.T) { + url := wsTestServer(t, func(ctx context.Context, c *websocket.Conn) { + if _, _, err := c.Read(ctx); err != nil { + return + } + _ = c.Write(ctx, websocket.MessageText, []byte( + `{"type":"conversation.item.input_audio_transcription.delta","delta":"hi"}`, + )) + _ = c.Write(ctx, websocket.MessageText, []byte( + `{"type":"error","error":{"message":"invalid API key sk-test-key"}}`, + )) + <-ctx.Done() + }) + + tr, err := NewOpenAIRealtimeTranscriber(OpenAIRealtimeConfig{APIKey: "sk-test-key", BaseURL: url}) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + chunks := make(chan []byte, 1) + defer close(chunks) + chunks <- make([]byte, 480) + var once sync.Once + _, ferr := tr.StreamTranscribe(ctx, chunks, func(string, bool) { + // Cancel synchronously from the partial callback, before the + // stream loop continues to the already-buffered error frame. + once.Do(cancel) + }) + if !errors.Is(ferr, context.Canceled) { + t.Fatalf("StreamTranscribe error = %v, want context.Canceled (cancel must win over a racing OpenAI error event)", ferr) + } +} diff --git a/internal/tui/dictation.go b/internal/tui/dictation.go index 3d96d82f6..348b72b13 100644 --- a/internal/tui/dictation.go +++ b/internal/tui/dictation.go @@ -61,6 +61,7 @@ type dictationController struct { browseVariants []dictation.ModelVariant // in-flight session state + sessionID int64 streaming bool recorder Recorder transcriber Transcriber @@ -101,13 +102,17 @@ type dictationController struct { } // dictationStartedMsg reports the outcome of Recorder.Start(). +// sessionID must match the dictation controller's current session so a late +// start result after cancel/reset cannot reset a newer recording. type dictationStartedMsg struct { - err error + sessionID int64 + err error } // dictationTranscribedMsg carries the final transcript of a batch (or a // completed streaming) recording. type dictationTranscribedMsg struct { + sessionID int64 text string err error submit bool @@ -125,6 +130,7 @@ func newDictationController(opts Options) dictationController { platform: dictation.DetectPlatform(), downloadRoot: opts.STTDownloadRoot, userConfigPath: opts.UserConfigPath, + sessionID: 1, } } @@ -262,7 +268,7 @@ func (m model) startDictation() (model, tea.Cmd) { if streaming { return m.startStreamingDictation() } - return m, startBatchRecordingCmd(rec) + return m, startBatchRecordingCmd(m.dictation.sessionID, rec) } // stopDictation ends a recording. For batch it triggers Stop()+Transcribe(); for @@ -276,7 +282,7 @@ func (m model) stopDictation() (model, tea.Cmd) { } return m, nil // final text arrives via the streaming command already running } - return m, transcribeBatchCmd(m.dictation.ctx, m.dictation.recorder, m.dictation.transcriber, m.dictation.cfg.AutoSubmitEnabled()) + return m, transcribeBatchCmd(m.dictation.sessionID, m.dictation.ctx, m.dictation.recorder, m.dictation.transcriber, m.dictation.cfg.AutoSubmitEnabled()) } // cancelDictation aborts an in-flight recording without transcribing. Bound to @@ -311,6 +317,7 @@ func (d dictationController) maxDuration() time.Duration { } func (d *dictationController) reset() { + d.sessionID++ d.phase = dictIdle d.recorder = nil d.transcriber = nil @@ -326,31 +333,36 @@ func (d *dictationController) reset() { // startBatchRecordingCmd starts the record-to-file capture off the UI goroutine // (Start spawns a subprocess and may briefly block or fail on a missing tool). -func startBatchRecordingCmd(rec Recorder) tea.Cmd { +func startBatchRecordingCmd(sessionID int64, rec Recorder) tea.Cmd { return func() tea.Msg { - return dictationStartedMsg{err: rec.Start()} + return dictationStartedMsg{sessionID: sessionID, err: rec.Start()} } } // transcribeBatchCmd stops the recording, transcribes the audio, and reports the // final text. Runs off the UI goroutine — Stop waits on the capture tool and // Transcribe does a network round-trip or a local exec. -func transcribeBatchCmd(ctx context.Context, rec Recorder, transcriber Transcriber, submit bool) tea.Cmd { +func transcribeBatchCmd(sessionID int64, ctx context.Context, rec Recorder, transcriber Transcriber, submit bool) tea.Cmd { if ctx == nil { ctx = context.Background() } return func() tea.Msg { audio, err := rec.Stop() if err != nil { - return dictationTranscribedMsg{err: err} + return dictationTranscribedMsg{sessionID: sessionID, err: err} } text, err := transcriber.Transcribe(ctx, audio) - return dictationTranscribedMsg{text: text, err: err, submit: submit} + return dictationTranscribedMsg{sessionID: sessionID, text: text, err: err, submit: submit} } } // handleDictationStarted transitions to recording (or reports a start failure). func (m model) handleDictationStarted(msg dictationStartedMsg) (model, tea.Cmd) { + if msg.sessionID != m.dictation.sessionID { + // Stale startup from a cancelled session must not reset or re-arm a + // newer recording that already owns the controller. + return m, nil + } if msg.err != nil { m = m.discardDictationRegion() m.dictation.reset() @@ -375,6 +387,9 @@ func (m model) handleDictationStarted(msg dictationStartedMsg) (model, tea.Cmd) // handleDictationTranscribed inserts the final transcript into the composer (or // submits it when stt.autoSubmit is on), then returns to idle. func (m model) handleDictationTranscribed(msg dictationTranscribedMsg) (tea.Model, tea.Cmd) { + if msg.sessionID != m.dictation.sessionID { + return m, nil + } streaming := m.dictation.streaming || msg.streaming m = m.commitDictationRegion() // A streaming session can end via a transcriber error (not just a user stop), so @@ -392,7 +407,17 @@ func (m model) handleDictationTranscribed(msg dictationTranscribedMsg) (tea.Mode } m.dictation.reset() - if msg.err != nil && !errors.Is(msg.err, context.Canceled) { + if errors.Is(msg.err, context.Canceled) { + // cancelDictation already discarded the live region, reset the + // session, and posted "Dictation cancelled." A streaming transcriber + // can still race a buffered event past that cancel and report back a + // nonempty compose()+context.Canceled; treat that as terminal here, + // before the auto-submit branch below, so Esc can never fall through + // to msg.submit and fire the composer's restored pre-existing text. + return m, nil + } + + if msg.err != nil { // A cloud auth failure (missing/invalid key) is fixable in place: reopen the // API-key prompt for the current provider so the user can paste a key and // retry, instead of hitting a dead-end "run zero auth" line. diff --git a/internal/tui/dictation_stream.go b/internal/tui/dictation_stream.go index 90b74c321..5e994eedd 100644 --- a/internal/tui/dictation_stream.go +++ b/internal/tui/dictation_stream.go @@ -14,8 +14,9 @@ import ( // mechanism agent text deltas use (§6). text is the cumulative best transcript // so far (not a delta); final marks a settled segment. type sttPartialMsg struct { - text string - final bool + sessionID int64 + text string + final bool } // startStreamingDictation begins continuous capture and launches the streaming @@ -38,6 +39,7 @@ func (m model) startStreamingDictation() (model, tea.Cmd) { } transcriber := m.dictation.transcriber submit := m.dictation.cfg.AutoSubmitEnabled() + sessionID := m.dictation.sessionID // Tap the audio: compute a mic level per chunk for the live waveform, then // forward the chunk to the transcriber. A small buffer keeps the tap from @@ -63,11 +65,11 @@ func (m model) startStreamingDictation() (model, tea.Cmd) { streamCmd := func() tea.Msg { onPartial := func(text string, final bool) { if sink != nil { - sink(sttPartialMsg{text: text, final: final}) + sink(sttPartialMsg{sessionID: sessionID, text: text, final: final}) } } text, err := transcriber.StreamTranscribe(ctx, tapped, onPartial) - return dictationTranscribedMsg{text: text, err: err, submit: submit, streaming: true} + return dictationTranscribedMsg{sessionID: sessionID, text: text, err: err, submit: submit, streaming: true} } // Streaming drives the waveform from real levels (no synthetic tick needed). return m, streamCmd @@ -77,6 +79,9 @@ func (m model) startStreamingDictation() (model, tea.Cmd) { // composer, replacing the previously-rendered live region wholesale so the text // builds up in place as the user keeps talking. func (m model) handleDictationPartial(msg sttPartialMsg) model { + if msg.sessionID != m.dictation.sessionID { + return m + } // Ignore stragglers that arrive after the session ended (cancel/final). if m.dictation.phase != dictRecording && m.dictation.phase != dictTranscribing { return m diff --git a/internal/tui/dictation_test.go b/internal/tui/dictation_test.go index 5101cbd64..508636168 100644 --- a/internal/tui/dictation_test.go +++ b/internal/tui/dictation_test.go @@ -86,9 +86,10 @@ func TestDictationNonAuthErrorDoesNotPrompt(t *testing.T) { func TestDictationStartedFailureResets(t *testing.T) { m := model{} + m.dictation.sessionID = 1 m.dictation.phase = dictStarting m.dictation.streaming = false - got, _ := m.handleDictationStarted(dictationStartedMsg{err: errors.New("mic busy")}) + got, _ := m.handleDictationStarted(dictationStartedMsg{sessionID: 1, err: errors.New("mic busy")}) if got.dictation.active() { t.Error("a start failure should reset to idle") } @@ -99,13 +100,48 @@ func TestDictationStartedFailureResets(t *testing.T) { func TestDictationStartedArmsRecording(t *testing.T) { m := model{} + m.dictation.sessionID = 1 m.dictation.phase = dictStarting - got, _ := m.handleDictationStarted(dictationStartedMsg{}) + got, _ := m.handleDictationStarted(dictationStartedMsg{sessionID: 1}) if got.dictation.phase != dictRecording { t.Error("a successful start should advance to recording") } } +// TestStaleDictationStartedIgnoredWhileLive pins the session gate on batch +// startup messages. A late Start() result after cancel must not reset() a +// newer recording (success would leave dictStarting with no recorder; failure +// would kill the live session). +func TestStaleDictationStartedIgnoredWhileLive(t *testing.T) { + // Active recording on session 2. + m := model{} + m.dictation.sessionID = 2 + m.dictation.phase = dictRecording + m.dictation.streaming = true + m.setComposerState(composerState{text: "live session text", cursor: len("live session text")}) + + // Stale failure from session 1 must not reset the live controller. + afterFail, _ := m.handleDictationStarted(dictationStartedMsg{ + sessionID: 1, + err: errors.New("mic busy from previous start"), + }) + if afterFail.dictation.phase != dictRecording || afterFail.dictation.sessionID != 2 { + t.Fatalf("stale start failure: phase=%v session=%d, want recording/2", afterFail.dictation.phase, afterFail.dictation.sessionID) + } + if transcriptHasText(afterFail, "mic busy from previous start") { + t.Fatal("stale start failure must not post an error notice") + } + + // Stale success from session 1 must not leave the live session stuck mid-start. + m2 := model{} + m2.dictation.sessionID = 2 + m2.dictation.phase = dictRecording + afterOK, _ := m2.handleDictationStarted(dictationStartedMsg{sessionID: 1}) + if afterOK.dictation.phase != dictRecording || afterOK.dictation.sessionID != 2 { + t.Fatalf("stale start success: phase=%v session=%d, want recording/2", afterOK.dictation.phase, afterOK.dictation.sessionID) + } +} + func TestDictationUnavailableShowsHint(t *testing.T) { m := model{} // no build factory next, _ := m.toggleDictation() @@ -136,6 +172,69 @@ func TestDictationStreamingPartialReplacesRegion(t *testing.T) { } } +func TestDictationCanceledStreamRaceDoesNotAutoSubmit(t *testing.T) { + // Esc can race an already-buffered realtime event: cancelDictation discards + // the live region and resets state synchronously, but the streaming + // goroutine's dictationTranscribedMsg (a nonempty compose() alongside + // context.Canceled) can still arrive afterward. With stt.autoSubmit on, + // that must never fall through to msg.submit and fire the composer's + // restored pre-existing text. + // Matching sessionID is required so this hits the context.Canceled branch + // rather than only the stale-session gate (session 0 after cancel would be + // dropped before that branch runs). + m := model{} + m.dictation.sessionID = 1 + m.setComposerState(composerState{text: "existing prompt", cursor: len("existing prompt")}) + m.dictation.phase = dictRecording + m.dictation.streaming = true + + m = m.handleDictationPartial(sttPartialMsg{sessionID: 1, text: "half-formed transcript"}) + if m.composer.text != "existing prompt half-formed transcript" { + t.Fatalf("partial did not render into composer: %q", m.composer.text) + } + + m, _ = m.cancelDictation() + if m.composer.text != "existing prompt" { + t.Fatalf("cancel should restore the pre-existing composer text, got %q", m.composer.text) + } + + // The already-buffered event's message arrives after the cancel above. + next, cmd := m.handleDictationTranscribed(dictationTranscribedMsg{ + sessionID: m.dictation.sessionID, + text: "half-formed transcript", + err: context.Canceled, + submit: true, + streaming: true, + }) + got := next.(model) + if got.composer.text != "existing prompt" { + t.Errorf("a raced cancellation must not auto-submit the restored composer text, got %q", got.composer.text) + } + if cmd != nil { + t.Error("a raced cancellation must not return a submit command") + } +} + +// TestStaleDictationPartialIgnoredWhileLive pins the session-ID gate on +// handleDictationPartial while a new session is still recording. Cancel-then- +// stale cases are already covered by the phase guard; this case needs a live +// phase so only the session check drops the ghost partial. +func TestStaleDictationPartialIgnoredWhileLive(t *testing.T) { + m := model{} + m.dictation.sessionID = 2 + m.dictation.phase = dictRecording + m.dictation.streaming = true + m.setComposerState(composerState{text: "user typed this", cursor: len("user typed this")}) + + got := m.handleDictationPartial(sttPartialMsg{ + sessionID: 1, + text: "ghost text from session 1", + }) + if got.composer.text != "user typed this" { + t.Fatalf("stale partial while live: got %q, want %q", got.composer.text, "user typed this") + } +} + func TestDictationCommitKeepsStreamedText(t *testing.T) { m := model{} m.setComposerState(composerState{text: "", cursor: 0}) @@ -302,3 +401,83 @@ func TestCurrentModelLabel(t *testing.T) { t.Errorf("no-model label = %q", got) } } + +func TestStaleDictationCompletionIgnoredAfterCancel(t *testing.T) { + m := model{} + m.setComposerState(composerState{text: "original composer text", cursor: 22}) + m.dictation.sessionID = 1 + m.dictation.phase = dictRecording + + m, _ = m.cancelDictation() + + if m.composer.text != "original composer text" { + t.Fatalf("composer text = %q, want 'original composer text'", m.composer.text) + } + + // Stale partial message from session 1 must be ignored + stalePartial := sttPartialMsg{ + sessionID: 1, + text: "stale partial that should be ignored", + } + mPartial := m.handleDictationPartial(stalePartial) + if mPartial.composer.text != "original composer text" { + t.Fatalf("composer text changed to %q after stale partial", mPartial.composer.text) + } + + staleMsg := dictationTranscribedMsg{ + sessionID: 1, + text: "stale text that should be ignored", + submit: true, + streaming: true, + } + afterStale, cmd := m.handleDictationTranscribed(staleMsg) + got := afterStale.(model) + + if got.composer.text != "original composer text" { + t.Fatalf("composer text changed to %q after stale completion", got.composer.text) + } + if cmd != nil { + t.Fatalf("expected no command for stale completion, got %v", cmd) + } +} + +func TestStaleDictationSessionZeroIgnoredAfterCancel(t *testing.T) { + m := model{} + m.setComposerState(composerState{text: "initial composer text", cursor: 21}) + m.dictation.sessionID = 0 + m.dictation.phase = dictRecording + + m, _ = m.cancelDictation() + + // Session 0 was reset to session 1 + if m.dictation.sessionID != 1 { + t.Fatalf("dictation.sessionID = %d after cancel, want 1", m.dictation.sessionID) + } + + // Stale partial from session 0 must be ignored + stalePartial := sttPartialMsg{ + sessionID: 0, + text: "stale partial from session 0", + } + mPartial := m.handleDictationPartial(stalePartial) + if mPartial.composer.text != "initial composer text" { + t.Fatalf("composer text changed to %q after stale session 0 partial", mPartial.composer.text) + } + + // Stale completion from session 0 must be ignored + staleMsg := dictationTranscribedMsg{ + sessionID: 0, + text: "stale completion from session 0", + submit: true, + streaming: true, + } + afterStale, cmd := m.handleDictationTranscribed(staleMsg) + got := afterStale.(model) + + if got.composer.text != "initial composer text" { + t.Fatalf("composer text changed to %q after stale session 0 completion", got.composer.text) + } + if cmd != nil { + t.Fatalf("expected no command for stale session 0 completion, got %v", cmd) + } +}