diff --git a/apps/sotto/internal/app/app_test.go b/apps/sotto/internal/app/app_test.go index 6f82069..ff73983 100644 --- a/apps/sotto/internal/app/app_test.go +++ b/apps/sotto/internal/app/app_test.go @@ -3,12 +3,18 @@ package app import ( "bytes" "context" + "errors" + "log/slog" "net" "os" "path/filepath" + "syscall" "testing" + "time" + "github.com/rbright/sotto/internal/fsm" "github.com/rbright/sotto/internal/ipc" + "github.com/rbright/sotto/internal/session" "github.com/stretchr/testify/require" ) @@ -201,6 +207,88 @@ func TestRunnerDevicesCommandDispatches(t *testing.T) { require.Contains(t, stderr.String(), "error:") } +func TestRunnerToggleOwnerPathReturnsErrorWhenCaptureStartupFails(t *testing.T) { + paths := setupRunnerEnv(t) + t.Setenv("PULSE_SERVER", "unix:/tmp/definitely-missing-pulse-server") + + var stdout bytes.Buffer + var stderr bytes.Buffer + runner := Runner{Stdout: &stdout, Stderr: &stderr} + + exitCode := runner.Execute(context.Background(), []string{"--config", paths.configPath, "toggle"}) + require.Equal(t, 1, exitCode) + require.Contains(t, stderr.String(), "error:") + + // owner path should clean up runtime socket on exit + _, statErr := os.Stat(filepath.Join(paths.runtimeDir, "sotto.sock")) + require.ErrorIs(t, statErr, os.ErrNotExist) +} + +func TestRunnerStatusFallsBackToIdleWhenServerStateEmpty(t *testing.T) { + paths := setupRunnerEnv(t) + + shutdown := startIPCServerForRunnerTest(t, filepath.Join(paths.runtimeDir, "sotto.sock"), func(_ context.Context, req ipc.Request) ipc.Response { + require.Equal(t, "status", req.Command) + return ipc.Response{OK: true, State: ""} + }) + defer shutdown() + + var stdout bytes.Buffer + var stderr bytes.Buffer + runner := Runner{Stdout: &stdout, Stderr: &stderr} + + exitCode := runner.Execute(context.Background(), []string{"--config", paths.configPath, "status"}) + require.Equal(t, 0, exitCode) + require.Equal(t, "idle\n", stdout.String()) + require.Empty(t, stderr.String()) +} + +func TestSocketErrorHelpers(t *testing.T) { + require.False(t, isSocketMissing(nil)) + require.False(t, isConnectionRefused(nil)) + + require.True(t, isSocketMissing(os.ErrNotExist)) + require.True(t, isSocketMissing(errors.New("dial unix /tmp/sotto.sock: no such file or directory"))) + require.False(t, isSocketMissing(errors.New("other error"))) + + require.True(t, isConnectionRefused(syscall.ECONNREFUSED)) + require.False(t, isConnectionRefused(errors.New("other error"))) +} + +func TestLogSessionResultWritesFailureAndSuccess(t *testing.T) { + var logBuf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logBuf, nil)) + + started := time.Now() + finished := started.Add(1500 * time.Millisecond) + + logSessionResult(logger, session.Result{ + State: fsm.StateIdle, + Cancelled: false, + StartedAt: started, + FinishedAt: finished, + AudioDevice: "Mic", + BytesCaptured: 123, + Transcript: "hello", + GRPCLatency: 20 * time.Millisecond, + }) + + require.Contains(t, logBuf.String(), "session complete") + require.Contains(t, logBuf.String(), "\"transcript_length\":5") + + logBuf.Reset() + logSessionResult(logger, session.Result{ + State: fsm.StateIdle, + StartedAt: started, + FinishedAt: finished, + Transcript: "", + Err: errors.New("boom"), + GRPCLatency: 2 * time.Millisecond, + }) + require.Contains(t, logBuf.String(), "session failed") + require.Contains(t, logBuf.String(), "boom") +} + type runnerPaths struct { configPath string runtimeDir string diff --git a/apps/sotto/internal/audio/pulse_test.go b/apps/sotto/internal/audio/pulse_test.go index 8ffa322..1efd442 100644 --- a/apps/sotto/internal/audio/pulse_test.go +++ b/apps/sotto/internal/audio/pulse_test.go @@ -1,6 +1,7 @@ package audio import ( + "context" "io" "reflect" "testing" @@ -59,6 +60,18 @@ func TestDeviceMatchesByIDAndDescription(t *testing.T) { require.False(t, deviceMatches(dev, "missing")) } +func TestListDevicesFailsWhenPulseUnavailable(t *testing.T) { + t.Setenv("PULSE_SERVER", "unix:/tmp/definitely-missing-pulse-server") + _, err := ListDevices(context.Background()) + require.Error(t, err) +} + +func TestSelectDeviceFailsWhenPulseUnavailable(t *testing.T) { + t.Setenv("PULSE_SERVER", "unix:/tmp/definitely-missing-pulse-server") + _, err := SelectDevice(context.Background(), "default", "default") + require.Error(t, err) +} + func TestSourceStateString(t *testing.T) { require.Equal(t, "running", sourceStateString(0)) require.Equal(t, "idle", sourceStateString(1)) @@ -136,6 +149,19 @@ func TestCaptureOnPCMReturnsEOFWhenStopped(t *testing.T) { require.Equal(t, int64(0), capture.BytesCaptured()) } +func TestCaptureDeviceAndCloseAlias(t *testing.T) { + capture := &Capture{ + device: Device{ID: "mic-1", Description: "Mic"}, + chunks: make(chan []byte, 1), + stopCh: make(chan struct{}), + } + require.Equal(t, "mic-1", capture.Device().ID) + + capture.Close() + _, ok := <-capture.Chunks() + require.False(t, ok) +} + type sourcePort struct { name string available uint32 diff --git a/apps/sotto/internal/config/parser_jsonc.go b/apps/sotto/internal/config/parser_jsonc.go index a117853..6050a6e 100644 --- a/apps/sotto/internal/config/parser_jsonc.go +++ b/apps/sotto/internal/config/parser_jsonc.go @@ -423,7 +423,7 @@ func isJSONWhitespace(ch byte) bool { } func ensureSingleJSONValue(decoder *json.Decoder) error { - var extra struct{} + var extra json.RawMessage err := decoder.Decode(&extra) if errors.Is(err, io.EOF) { return nil diff --git a/apps/sotto/internal/config/parser_jsonc_test.go b/apps/sotto/internal/config/parser_jsonc_test.go new file mode 100644 index 0000000..add8328 --- /dev/null +++ b/apps/sotto/internal/config/parser_jsonc_test.go @@ -0,0 +1,142 @@ +package config + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNormalizeJSONCRemovesCommentsAndTrailingCommas(t *testing.T) { + input := ` +{ + // line comment + "items": [ + "one", /* block comment */ + "two", + ], + "nested": { + "enabled": true, + }, +} +` + + normalized, err := normalizeJSONC(input) + require.NoError(t, err) + require.NotContains(t, normalized, "//") + require.NotContains(t, normalized, "/*") + require.NotContains(t, normalized, ",]") + require.NotContains(t, normalized, ",}") +} + +func TestNormalizeJSONCRetainsCommentLikeTextInsideStrings(t *testing.T) { + input := `{"value":"contains // and /* comment-like */ text",}` + normalized, err := normalizeJSONC(input) + require.NoError(t, err) + require.Contains(t, normalized, "// and /* comment-like */") +} + +func TestNormalizeJSONCUnterminatedBlockCommentFails(t *testing.T) { + _, err := normalizeJSONC("{ /* unterminated ") + require.Error(t, err) + require.Contains(t, err.Error(), "unterminated block comment") +} + +func TestEnsureSingleJSONValueRejectsExtraPayload(t *testing.T) { + decoder := json.NewDecoder(strings.NewReader(`{"one":1}{"two":2}`)) + var payload map[string]any + require.NoError(t, decoder.Decode(&payload)) + + err := ensureSingleJSONValue(decoder) + require.Error(t, err) + require.Contains(t, err.Error(), "multiple JSON values") +} + +func TestOffsetToLineCol(t *testing.T) { + content := "line1\nline2\nline3" + line, col := offsetToLineCol(content, 1) + require.Equal(t, 1, line) + require.Equal(t, 1, col) + + line, col = offsetToLineCol(content, 8) // line2, col2 + require.Equal(t, 2, line) + require.Equal(t, 2, col) + + line, col = offsetToLineCol(content, 999) + require.Equal(t, 3, line) + require.Equal(t, 5, col) +} + +func TestJSONCStringListUnmarshal(t *testing.T) { + var list jsoncStringList + require.NoError(t, list.UnmarshalJSON([]byte(`["a","b"]`))) + require.Equal(t, []string{"a", "b"}, []string(list)) + + require.NoError(t, list.UnmarshalJSON([]byte(`"a, b, , c"`))) + require.Equal(t, []string{"a", "b", "c"}, []string(list)) + + err := list.UnmarshalJSON([]byte(`123`)) + require.Error(t, err) + require.Contains(t, err.Error(), "expected string array") +} + +func TestParseJSONCRejectsInvalidCommandArgv(t *testing.T) { + _, _, err := parseJSONC(`{"clipboard_cmd":"unterminated ' quote"}`, Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "invalid clipboard_cmd") + + _, _, err = parseJSONC(`{"paste_cmd":"unterminated ' quote"}`, Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "invalid paste_cmd") +} + +func TestParseJSONCVocabRejectsEmptySetName(t *testing.T) { + _, _, err := parseJSONC(`{"vocab":{"sets":{" ":{"phrases":["x"]}}}}`, Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "empty set name") +} + +func TestParseJSONCTrimsIndicatorAndPasteFields(t *testing.T) { + cfg, _, err := parseJSONC(`{ + "paste": {"shortcut": " CTRL,V "}, + "indicator": { + "backend": " desktop ", + "desktop_app_name": " sotto-indicator " + } +}`, Default()) + require.NoError(t, err) + require.Equal(t, "CTRL,V", cfg.Paste.Shortcut) + require.Equal(t, "desktop", cfg.Indicator.Backend) + require.Equal(t, "sotto-indicator", cfg.Indicator.DesktopAppName) +} + +func TestParseJSONCRejectsMultipleTopLevelValues(t *testing.T) { + _, _, err := parseJSONC(`{"paste":{"enable":false}}{"paste":{"enable":true}}`, Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "multiple JSON values") +} + +func TestParseJSONCTypeErrorIncludesLocation(t *testing.T) { + _, _, err := parseJSONC(`{ + "riva": {"grpc": 123} +}`, Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "line") + require.Contains(t, err.Error(), "column") +} + +func TestParseJSONCVocabGlobalSupportsCommaString(t *testing.T) { + cfg, _, err := parseJSONC(`{ + "vocab": { + "global": "one, two, , three", + "sets": { + "one": {"phrases": ["one"]}, + "two": {"phrases": ["two"]}, + "three": {"phrases": ["three"]} + } + } +}`, Default()) + require.NoError(t, err) + require.Equal(t, []string{"one", "two", "three"}, cfg.Vocab.GlobalSets) +} diff --git a/apps/sotto/internal/doctor/doctor_test.go b/apps/sotto/internal/doctor/doctor_test.go index 428eafb..c94531b 100644 --- a/apps/sotto/internal/doctor/doctor_test.go +++ b/apps/sotto/internal/doctor/doctor_test.go @@ -131,3 +131,65 @@ func TestCheckAudioSelectionFailureWithInvalidPulseServer(t *testing.T) { require.False(t, check.Pass) require.Contains(t, check.Name, "audio.device") } + +func TestReportOKAllPassing(t *testing.T) { + report := Report{Checks: []Check{{Name: "one", Pass: true}, {Name: "two", Pass: true}}} + require.True(t, report.OK()) +} + +func TestRunUsesPasteCmdOverrideCheck(t *testing.T) { + binDir := t.TempDir() + fakePaste := filepath.Join(binDir, "fake-paste") + require.NoError(t, os.WriteFile(fakePaste, []byte("#!/usr/bin/env sh\nexit 0\n"), 0o755)) + t.Setenv("PATH", binDir+":"+os.Getenv("PATH")) + t.Setenv("PULSE_SERVER", "unix:/tmp/definitely-missing-pulse-server") + t.Setenv("XDG_SESSION_TYPE", "wayland") + t.Setenv("HYPRLAND_INSTANCE_SIGNATURE", "abc123") + + cfg := config.Default() + cfg.Paste.Enable = true + cfg.PasteCmd = config.CommandConfig{Raw: fakePaste, Argv: []string{"fake-paste"}} + cfg.RivaHTTP = "" + + report := Run(config.Loaded{Path: "/tmp/config.jsonc", Config: cfg}) + require.NotEmpty(t, report.Checks) + + var sawPasteCmd, sawHypr bool + for _, check := range report.Checks { + if check.Name == "fake-paste" { + sawPasteCmd = true + } + if check.Name == "hyprctl" { + sawHypr = true + } + } + require.True(t, sawPasteCmd) + require.False(t, sawHypr) +} + +func TestRunUsesHyprctlWhenPasteCmdUnset(t *testing.T) { + binDir := t.TempDir() + fakeHypr := filepath.Join(binDir, "hyprctl") + require.NoError(t, os.WriteFile(fakeHypr, []byte("#!/usr/bin/env sh\nexit 0\n"), 0o755)) + t.Setenv("PATH", binDir+":"+os.Getenv("PATH")) + t.Setenv("PULSE_SERVER", "unix:/tmp/definitely-missing-pulse-server") + t.Setenv("XDG_SESSION_TYPE", "wayland") + t.Setenv("HYPRLAND_INSTANCE_SIGNATURE", "abc123") + + cfg := config.Default() + cfg.Paste.Enable = true + cfg.PasteCmd = config.CommandConfig{} + cfg.RivaHTTP = "" + + report := Run(config.Loaded{Path: "/tmp/config.jsonc", Config: cfg}) + require.NotEmpty(t, report.Checks) + + var sawHypr bool + for _, check := range report.Checks { + if check.Name == "hyprctl" { + sawHypr = true + break + } + } + require.True(t, sawHypr) +} diff --git a/apps/sotto/internal/pipeline/transcriber.go b/apps/sotto/internal/pipeline/transcriber.go index bdff4f5..7d26fc9 100644 --- a/apps/sotto/internal/pipeline/transcriber.go +++ b/apps/sotto/internal/pipeline/transcriber.go @@ -19,6 +19,21 @@ import ( "github.com/rbright/sotto/internal/transcript" ) +// captureClient is the audio-capture contract needed by the transcriber. +type captureClient interface { + Stop() error + Chunks() <-chan []byte + BytesCaptured() int64 + RawPCM() []byte +} + +// streamClient is the ASR-stream contract needed by the transcriber. +type streamClient interface { + SendAudio([]byte) error + CloseAndCollect(context.Context) ([]string, time.Duration, error) + Cancel() error +} + // Transcriber owns one end-to-end capture -> ASR -> transcript pipeline instance. type Transcriber struct { cfg config.Config @@ -28,17 +43,33 @@ type Transcriber struct { started bool selection audio.Selection - capture *audio.Capture - stream *riva.Stream + capture captureClient + stream streamClient sendErrCh chan error + selectDevice func(context.Context, string, string) (audio.Selection, error) + startCapture func(context.Context, audio.Device) (captureClient, error) + dialStream func(context.Context, riva.StreamConfig) (streamClient, error) + debugGRPCFile *os.File } // NewTranscriber constructs a pipeline transcriber from runtime config. func NewTranscriber(cfg config.Config, logger *slog.Logger) *Transcriber { - return &Transcriber{cfg: cfg, logger: logger} + return &Transcriber{ + cfg: cfg, + logger: logger, + selectDevice: func(ctx context.Context, input string, fallback string) (audio.Selection, error) { + return audio.SelectDevice(ctx, input, fallback) + }, + startCapture: func(ctx context.Context, device audio.Device) (captureClient, error) { + return audio.StartCapture(ctx, device) + }, + dialStream: func(ctx context.Context, cfg riva.StreamConfig) (streamClient, error) { + return riva.DialStream(ctx, cfg) + }, + } } // Start resolves device selection, opens Riva stream, and starts audio capture. @@ -50,7 +81,7 @@ func (t *Transcriber) Start(ctx context.Context) error { return fmt.Errorf("transcriber already started") } - selection, err := audio.SelectDevice(ctx, t.cfg.Audio.Input, t.cfg.Audio.Fallback) + selection, err := t.selectDevice(ctx, t.cfg.Audio.Input, t.cfg.Audio.Fallback) if err != nil { return err } @@ -77,7 +108,7 @@ func (t *Transcriber) Start(ctx context.Context) error { rivaPhrases = append(rivaPhrases, riva.SpeechPhrase{Phrase: phrase.Phrase, Boost: phrase.Boost}) } - stream, err := riva.DialStream(ctx, riva.StreamConfig{ + stream, err := t.dialStream(ctx, riva.StreamConfig{ Endpoint: t.cfg.RivaGRPC, LanguageCode: t.cfg.ASR.LanguageCode, Model: t.cfg.ASR.Model, @@ -97,7 +128,7 @@ func (t *Transcriber) Start(ctx context.Context) error { } t.stream = stream - capture, err := audio.StartCapture(ctx, selection.Device) + capture, err := t.startCapture(ctx, selection.Device) if err != nil { _ = stream.Cancel() t.closeDebugArtifactsLocked() diff --git a/apps/sotto/internal/pipeline/transcriber_test.go b/apps/sotto/internal/pipeline/transcriber_test.go index 295fa91..71c2012 100644 --- a/apps/sotto/internal/pipeline/transcriber_test.go +++ b/apps/sotto/internal/pipeline/transcriber_test.go @@ -3,12 +3,15 @@ package pipeline import ( "context" "encoding/binary" + "errors" "os" "path/filepath" "testing" + "time" "github.com/rbright/sotto/internal/audio" "github.com/rbright/sotto/internal/config" + "github.com/rbright/sotto/internal/riva" "github.com/rbright/sotto/internal/session" "github.com/stretchr/testify/require" ) @@ -179,3 +182,257 @@ func TestCloseDebugArtifactsLockedClosesFileWhileMutexHeld(t *testing.T) { require.Error(t, err) require.Nil(t, transcriber.debugGRPCFile) } + +func TestStartWiresDependenciesAndBootsSendLoop(t *testing.T) { + cfg := config.Default() + transcriber := NewTranscriber(cfg, nil) + + chunks := make(chan []byte) + close(chunks) + capture := &fakeCapture{chunks: chunks} + stream := &fakeStream{} + + transcriber.selectDevice = func(context.Context, string, string) (audio.Selection, error) { + return audio.Selection{Device: audio.Device{ID: "mic-1", Description: "Mic"}}, nil + } + transcriber.dialStream = func(context.Context, riva.StreamConfig) (streamClient, error) { + return stream, nil + } + transcriber.startCapture = func(context.Context, audio.Device) (captureClient, error) { + return capture, nil + } + + err := transcriber.Start(context.Background()) + require.NoError(t, err) + require.True(t, transcriber.started) + require.Equal(t, "mic-1", transcriber.selection.Device.ID) + require.NotNil(t, transcriber.sendErrCh) + + require.NoError(t, transcriber.Cancel(context.Background())) +} + +func TestStartFailsOnSpeechPhraseBuildError(t *testing.T) { + cfg := config.Default() + cfg.Vocab.GlobalSets = []string{"missing"} + + transcriber := NewTranscriber(cfg, nil) + transcriber.selectDevice = func(context.Context, string, string) (audio.Selection, error) { + return audio.Selection{Device: audio.Device{ID: "mic-1", Description: "Mic"}}, nil + } + transcriber.dialStream = func(context.Context, riva.StreamConfig) (streamClient, error) { + t.Fatal("dialStream should not be called when speech phrase build fails") + return nil, nil + } + transcriber.startCapture = func(context.Context, audio.Device) (captureClient, error) { + t.Fatal("startCapture should not be called when speech phrase build fails") + return nil, nil + } + + err := transcriber.Start(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "build speech contexts") + require.False(t, transcriber.started) +} + +func TestStopAndTranscribeSuccessPath(t *testing.T) { + cfg := config.Default() + cfg.Transcript.TrailingSpace = true + + capture := &fakeCapture{ + chunks: make(chan []byte), + raw: []byte{1, 2, 3, 4}, + bytes: 4096, + } + close(capture.chunks) + + stream := &fakeStream{ + closeSegments: []string{"hello", "world"}, + closeLatency: 12 * time.Millisecond, + } + + transcriber := NewTranscriber(cfg, nil) + transcriber.started = true + transcriber.selection = audio.Selection{Device: audio.Device{ID: "mic-1", Description: "Mic"}} + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + transcriber.sendErrCh <- nil + + result, err := transcriber.StopAndTranscribe(context.Background()) + require.NoError(t, err) + require.Equal(t, "hello world ", result.Transcript) + require.Equal(t, "Mic (mic-1)", result.AudioDevice) + require.Equal(t, int64(4096), result.BytesCaptured) + require.Equal(t, 12*time.Millisecond, result.GRPCLatency) + require.True(t, capture.stopCalled) + require.False(t, transcriber.started) + require.Nil(t, transcriber.capture) + require.Nil(t, transcriber.stream) +} + +func TestStopAndTranscribeSendErrorCancelsStream(t *testing.T) { + capture := &fakeCapture{ + chunks: make(chan []byte), + raw: []byte{1, 2, 3, 4}, + bytes: 2048, + } + close(capture.chunks) + stream := &fakeStream{} + + transcriber := NewTranscriber(config.Default(), nil) + transcriber.started = true + transcriber.selection = audio.Selection{Device: audio.Device{ID: "mic-1", Description: "Mic"}} + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + transcriber.sendErrCh <- errors.New("send failed") + + result, err := transcriber.StopAndTranscribe(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "send audio stream") + require.Equal(t, "Mic (mic-1)", result.AudioDevice) + require.True(t, stream.cancelCalled) +} + +func TestStopAndTranscribeCollectErrorIncludesLatency(t *testing.T) { + capture := &fakeCapture{ + chunks: make(chan []byte), + raw: []byte{1, 2, 3, 4}, + bytes: 1024, + } + close(capture.chunks) + + stream := &fakeStream{closeErr: errors.New("close failed"), closeLatency: 77 * time.Millisecond} + + transcriber := NewTranscriber(config.Default(), nil) + transcriber.started = true + transcriber.selection = audio.Selection{Device: audio.Device{ID: "mic-1", Description: "Mic"}} + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + transcriber.sendErrCh <- nil + + result, err := transcriber.StopAndTranscribe(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "collect final transcript") + require.Equal(t, 77*time.Millisecond, result.GRPCLatency) +} + +func TestCancelStopsCaptureAndStreamAndResetsState(t *testing.T) { + capture := &fakeCapture{chunks: make(chan []byte), raw: []byte{1}, bytes: 1} + close(capture.chunks) + stream := &fakeStream{} + + transcriber := NewTranscriber(config.Default(), nil) + transcriber.started = true + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + + err := transcriber.Cancel(context.Background()) + require.NoError(t, err) + require.True(t, capture.stopCalled) + require.True(t, stream.cancelCalled) + require.False(t, transcriber.started) + require.Nil(t, transcriber.capture) + require.Nil(t, transcriber.stream) + require.Nil(t, transcriber.sendErrCh) +} + +func TestSendLoopForwardsChunksAndSignalsNil(t *testing.T) { + chunks := make(chan []byte, 4) + chunks <- []byte{1, 2, 3} + chunks <- []byte{} + chunks <- []byte{4, 5} + close(chunks) + + capture := &fakeCapture{chunks: chunks} + stream := &fakeStream{} + transcriber := NewTranscriber(config.Default(), nil) + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + + transcriber.sendLoop() + + err := <-transcriber.sendErrCh + require.NoError(t, err) + require.Equal(t, 2, len(stream.sendChunks)) + require.Equal(t, []byte{1, 2, 3}, stream.sendChunks[0]) + require.Equal(t, []byte{4, 5}, stream.sendChunks[1]) +} + +func TestSendLoopStopsCaptureOnSendError(t *testing.T) { + chunks := make(chan []byte, 2) + chunks <- []byte{1, 2, 3} + close(chunks) + + capture := &fakeCapture{chunks: chunks} + stream := &fakeStream{sendErr: errors.New("boom")} + transcriber := NewTranscriber(config.Default(), nil) + transcriber.capture = capture + transcriber.stream = stream + transcriber.sendErrCh = make(chan error, 1) + + transcriber.sendLoop() + + err := <-transcriber.sendErrCh + require.Error(t, err) + require.Contains(t, err.Error(), "boom") + require.True(t, capture.stopCalled) +} + +type fakeCapture struct { + chunks chan []byte + stopErr error + raw []byte + bytes int64 + stopCalled bool +} + +func (f *fakeCapture) Stop() error { + f.stopCalled = true + return f.stopErr +} + +func (f *fakeCapture) Chunks() <-chan []byte { return f.chunks } + +func (f *fakeCapture) BytesCaptured() int64 { return f.bytes } + +func (f *fakeCapture) RawPCM() []byte { + out := make([]byte, len(f.raw)) + copy(out, f.raw) + return out +} + +type fakeStream struct { + sendErr error + closeErr error + closeSegments []string + closeLatency time.Duration + cancelCalled bool + sendChunks [][]byte +} + +func (f *fakeStream) SendAudio(chunk []byte) error { + if f.sendErr != nil { + return f.sendErr + } + copyChunk := make([]byte, len(chunk)) + copy(copyChunk, chunk) + f.sendChunks = append(f.sendChunks, copyChunk) + return nil +} + +func (f *fakeStream) CloseAndCollect(context.Context) ([]string, time.Duration, error) { + if f.closeErr != nil { + return nil, f.closeLatency, f.closeErr + } + segments := append([]string(nil), f.closeSegments...) + return segments, f.closeLatency, nil +} + +func (f *fakeStream) Cancel() error { + f.cancelCalled = true + return nil +} diff --git a/apps/sotto/internal/riva/client.go b/apps/sotto/internal/riva/client.go index bcc272b..6274366 100644 --- a/apps/sotto/internal/riva/client.go +++ b/apps/sotto/internal/riva/client.go @@ -42,8 +42,9 @@ type Stream struct { recvDone chan struct{} mu sync.Mutex - segments []string // committed transcript segments (final and high-confidence boundary-committed interim) + segments []string // committed transcript segments (final results and sealed interim chains) lastInterim string + lastInterimAge int lastInterimStability float32 recvErr error closedSend bool diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index c7afc9e..cd0e9a4 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -26,6 +26,14 @@ func TestCollectSegmentsFallsBackToInterim(t *testing.T) { require.Equal(t, []string{"tentative words"}, got) } +func TestCollectSegmentsMergesTrailingInterimWithCommittedSegments(t *testing.T) { + got := collectSegments([]string{"hello world"}, "hello world and beyond") + require.Equal(t, []string{"hello world and beyond"}, got) + + got = collectSegments([]string{"hello world"}, "hello") + require.Equal(t, []string{"hello world"}, got) +} + func TestRecordResponseTracksInterimThenFinal(t *testing.T) { s := &Stream{} @@ -37,6 +45,7 @@ func TestRecordResponseTracksInterimThenFinal(t *testing.T) { }) require.Equal(t, "hello wor", s.lastInterim) + require.Equal(t, 1, s.lastInterimAge) require.Empty(t, s.segments) s.recordResponse(&asrpb.StreamingRecognizeResponse{ @@ -47,6 +56,7 @@ func TestRecordResponseTracksInterimThenFinal(t *testing.T) { }) require.Empty(t, s.lastInterim) + require.Equal(t, 0, s.lastInterimAge) require.Equal(t, []string{"hello world"}, s.segments) } @@ -72,7 +82,7 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } -func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testing.T) { +func TestRecordResponseCommitsStableSingleInterimOnDivergence(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ @@ -96,13 +106,61 @@ func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testin require.Equal(t, []string{"first phrase", "second phrase"}, segments) } +func TestRecordResponseCommitsInterimChainOnDivergence(t *testing.T) { + s := &Stream{} + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase extended"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "second phrase"}}, + }}, + }) + + require.Equal(t, []string{"first phrase extended"}, s.segments) + require.Equal(t, "second phrase", s.lastInterim) + require.Equal(t, 1, s.lastInterimAge) + segments := collectSegments(s.segments, s.lastInterim) + require.Equal(t, []string{"first phrase extended", "second phrase"}, segments) +} + +func TestRecordResponseBuildsMultipleSegmentsAcrossLongInterimStream(t *testing.T) { + s := &Stream{} + + responses := []*asrpb.StreamingRecognizeResponse{ + {Results: []*asrpb.StreamingRecognitionResult{{IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "alpha"}}}}}, + {Results: []*asrpb.StreamingRecognitionResult{{IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "alpha one"}}}}}, + {Results: []*asrpb.StreamingRecognitionResult{{IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "beta"}}}}}, + {Results: []*asrpb.StreamingRecognitionResult{{IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "beta two"}}}}}, + {Results: []*asrpb.StreamingRecognitionResult{{IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "gamma"}}}}}, + } + for _, resp := range responses { + s.recordResponse(resp) + } + + segments := collectSegments(s.segments, s.lastInterim) + require.Equal(t, []string{"alpha one", "beta two", "gamma"}, segments) +} + func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.05, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "stale words"}}, }}, }) @@ -110,7 +168,6 @@ func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.30, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "hello world"}}, }}, }) @@ -132,7 +189,6 @@ func TestRecordResponseTreatsSuffixCorrectionAsContinuation(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.95, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "replace reply replied on the review thread with details"}}, }}, }) @@ -140,7 +196,6 @@ func TestRecordResponseTreatsSuffixCorrectionAsContinuation(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.95, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "replied on the review thread with details"}}, }}, }) @@ -167,19 +222,33 @@ func TestAppendSegmentDedupAndPrefixMerge(t *testing.T) { require.Equal(t, []string{"hello world", "new sentence"}, segments) } -func TestCleanSegmentAndInterimContinuation(t *testing.T) { +func TestInterimHelpers(t *testing.T) { require.Equal(t, "hello world", cleanSegment(" hello\n world ")) require.Empty(t, cleanSegment(" \n\t")) - require.True(t, isInterimContinuation("hello", "hello world")) - require.True(t, isInterimContinuation("hello world", "hello")) - require.True(t, isInterimContinuation("replace reply replied on thread", "replied on thread")) - require.False(t, isInterimContinuation("first phrase", "second phrase")) + continuationCases := []struct { + name string + previous string + current string + want bool + }{ + {name: "prefix extension", previous: "hello", current: "hello world", want: true}, + {name: "suffix correction", previous: "replace reply replied on thread", current: "replied on thread", want: true}, + {name: "shared prefix majority", previous: "one two three", current: "one two four", want: true}, + {name: "shared suffix majority", previous: "noise at start hello world", current: "hello world", want: true}, + {name: "divergent phrases", previous: "first phrase", current: "second phrase", want: false}, + } + for _, tc := range continuationCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isInterimContinuation(tc.previous, tc.current)) + }) + } - require.False(t, shouldCommitPriorInterimOnDivergence("first phrase", 0.2, "second phrase")) - require.True(t, shouldCommitPriorInterimOnDivergence("first phrase", 0.9, "second phrase")) - require.True(t, shouldCommitPriorInterimOnDivergence("Done.", 0.1, "new sentence")) - require.False(t, shouldCommitPriorInterimOnDivergence("replace reply replied on thread", 0.95, "replied on thread")) + require.False(t, shouldCommitInterimBoundary("", 5, 0.9)) + require.False(t, shouldCommitInterimBoundary("first phrase", 1, 0.1)) + require.True(t, shouldCommitInterimBoundary("first phrase", 2, 0.1)) + require.True(t, shouldCommitInterimBoundary("first phrase", 1, 0.9)) + require.True(t, shouldCommitInterimBoundary("done.", 1, 0.0)) } func TestDialStreamEndToEndWithDebugSinkAndSpeechContexts(t *testing.T) { diff --git a/apps/sotto/internal/riva/stream_receive.go b/apps/sotto/internal/riva/stream_receive.go index 769e642..e7801fe 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -53,14 +53,25 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { if result.GetIsFinal() { s.segments = appendSegment(s.segments, transcript) s.lastInterim = "" + s.lastInterimAge = 0 s.lastInterimStability = 0 continue } - if shouldCommitPriorInterimOnDivergence(s.lastInterim, s.lastInterimStability, transcript) { - s.segments = appendSegment(s.segments, s.lastInterim) + if s.lastInterim != "" { + if isInterimContinuation(s.lastInterim, transcript) { + s.lastInterim = transcript + s.lastInterimAge++ + s.lastInterimStability = result.GetStability() + continue + } + if shouldCommitInterimBoundary(s.lastInterim, s.lastInterimAge, s.lastInterimStability) { + s.segments = appendSegment(s.segments, s.lastInterim) + } } + s.lastInterim = transcript + s.lastInterimAge = 1 s.lastInterimStability = result.GetStability() } } diff --git a/apps/sotto/internal/riva/transcript_segments.go b/apps/sotto/internal/riva/transcript_segments.go index ea7efb2..17f0d48 100644 --- a/apps/sotto/internal/riva/transcript_segments.go +++ b/apps/sotto/internal/riva/transcript_segments.go @@ -2,7 +2,10 @@ package riva import "strings" -const stableInterimBoundaryThreshold = 0.85 +const ( + minInterimChainUpdates = 2 + stableInterimBoundaryThreshold = 0.85 +) // collectSegments appends a valid trailing interim segment when needed. func collectSegments(committedSegments []string, lastInterim string) []string { @@ -37,7 +40,8 @@ func appendSegment(segments []string, transcript string) []string { } } -// isInterimContinuation decides whether an interim update extends prior speech. +// isInterimContinuation reports whether the new interim looks like a rewrite or +// extension of the prior interim hypothesis. func isInterimContinuation(previous string, current string) bool { previous = cleanSegment(previous) current = cleanSegment(current) @@ -64,43 +68,38 @@ func isInterimContinuation(previous string, current string) bool { return true } - commonPrefix := commonPrefixWords(prevWords, currWords) - if commonPrefix*2 >= shorter { + if commonPrefixWords(prevWords, currWords)*2 >= shorter { return true } - - commonSuffix := commonSuffixWords(prevWords, currWords) - if shorter >= 3 && commonSuffix*2 >= shorter { + if shorter >= 3 && commonSuffixWords(prevWords, currWords)*2 >= shorter { return true } return false } -// shouldCommitPriorInterimOnDivergence decides whether to preserve prior interim -// text when a new interim hypothesis diverges. -func shouldCommitPriorInterimOnDivergence(previous string, previousStability float32, current string) bool { +// shouldCommitInterimBoundary returns true when a divergent interim chain looks +// established enough to preserve as a committed segment. +func shouldCommitInterimBoundary(previous string, chainUpdates int, stability float32) bool { previous = cleanSegment(previous) - current = cleanSegment(current) - if previous == "" || current == "" { + if previous == "" { return false } - if isInterimContinuation(previous, current) { - return false + if chainUpdates >= minInterimChainUpdates { + return true } - if previousStability >= stableInterimBoundaryThreshold { + if stability >= stableInterimBoundaryThreshold { return true } return endsWithSentencePunctuation(previous) } -// endsWithSentencePunctuation reports whether transcript looks sentence-complete. -func endsWithSentencePunctuation(transcript string) bool { - transcript = strings.TrimSpace(transcript) - if transcript == "" { +func endsWithSentencePunctuation(text string) bool { + text = strings.TrimSpace(text) + if text == "" { return false } - switch transcript[len(transcript)-1] { + switch text[len(text)-1] { case '.', '!', '?': return true default: