From 439c198021cc438e85833d2d0db86db6b854f4e3 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Sun, 22 Feb 2026 20:49:04 -0700 Subject: [PATCH 1/6] fix(riva): simplify interim handling to latest-per-toggle Remove interim divergence heuristics and keep session behavior strict: - commit only final ASR segments during an active toggle session - keep a single latest interim snapshot only as end-of-session fallback - no interim pre-commit or boundary inference across updates This aligns runtime behavior with toggle boundaries and avoids stale-prefix commit logic. Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client.go | 13 +-- apps/sotto/internal/riva/client_test.go | 18 +-- apps/sotto/internal/riva/stream_receive.go | 6 +- .../internal/riva/transcript_segments.go | 105 ------------------ 4 files changed, 11 insertions(+), 131 deletions(-) diff --git a/apps/sotto/internal/riva/client.go b/apps/sotto/internal/riva/client.go index bcc272b..44f844a 100644 --- a/apps/sotto/internal/riva/client.go +++ b/apps/sotto/internal/riva/client.go @@ -41,13 +41,12 @@ type Stream struct { recvDone chan struct{} - mu sync.Mutex - segments []string // committed transcript segments (final and high-confidence boundary-committed interim) - lastInterim string - lastInterimStability float32 - recvErr error - closedSend bool - debugSinkJSON io.Writer + mu sync.Mutex + segments []string // committed transcript segments (final results only) + lastInterim string + recvErr error + closedSend bool + debugSinkJSON io.Writer } // DialStream establishes a stream, sends config, and starts the receive loop. diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index c7afc9e..a9fc1be 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -72,7 +72,7 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } -func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testing.T) { +func TestRecordResponseDoesNotCommitDivergentInterimBeforeFinal(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ @@ -91,9 +91,9 @@ func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testin }}, }) - require.Equal(t, []string{"first phrase"}, s.segments) + require.Empty(t, s.segments) segments := collectSegments(s.segments, s.lastInterim) - require.Equal(t, []string{"first phrase", "second phrase"}, segments) + require.Equal(t, []string{"second phrase"}, segments) } func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { @@ -167,19 +167,9 @@ func TestAppendSegmentDedupAndPrefixMerge(t *testing.T) { require.Equal(t, []string{"hello world", "new sentence"}, segments) } -func TestCleanSegmentAndInterimContinuation(t *testing.T) { +func TestCleanSegment(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")) - - 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")) } 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..8c66834 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -53,14 +53,10 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { if result.GetIsFinal() { s.segments = appendSegment(s.segments, transcript) s.lastInterim = "" - s.lastInterimStability = 0 continue } - if shouldCommitPriorInterimOnDivergence(s.lastInterim, s.lastInterimStability, transcript) { - s.segments = appendSegment(s.segments, s.lastInterim) - } + // Keep only the latest interim hypothesis within a toggle session. s.lastInterim = transcript - s.lastInterimStability = result.GetStability() } } diff --git a/apps/sotto/internal/riva/transcript_segments.go b/apps/sotto/internal/riva/transcript_segments.go index ea7efb2..2020752 100644 --- a/apps/sotto/internal/riva/transcript_segments.go +++ b/apps/sotto/internal/riva/transcript_segments.go @@ -2,8 +2,6 @@ package riva import "strings" -const stableInterimBoundaryThreshold = 0.85 - // collectSegments appends a valid trailing interim segment when needed. func collectSegments(committedSegments []string, lastInterim string) []string { segments := append([]string(nil), committedSegments...) @@ -37,109 +35,6 @@ func appendSegment(segments []string, transcript string) []string { } } -// isInterimContinuation decides whether an interim update extends prior speech. -func isInterimContinuation(previous string, current string) bool { - previous = cleanSegment(previous) - current = cleanSegment(current) - if previous == "" || current == "" { - return true - } - if previous == current { - return true - } - if strings.HasPrefix(current, previous) || strings.HasPrefix(previous, current) { - return true - } - if strings.HasSuffix(current, previous) || strings.HasSuffix(previous, current) { - return true - } - - prevWords := strings.Fields(previous) - currWords := strings.Fields(current) - shorter := len(prevWords) - if len(currWords) < shorter { - shorter = len(currWords) - } - if shorter == 0 { - return true - } - - commonPrefix := commonPrefixWords(prevWords, currWords) - if commonPrefix*2 >= shorter { - return true - } - - commonSuffix := commonSuffixWords(prevWords, currWords) - if shorter >= 3 && commonSuffix*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 { - previous = cleanSegment(previous) - current = cleanSegment(current) - if previous == "" || current == "" { - return false - } - if isInterimContinuation(previous, current) { - return false - } - if previousStability >= stableInterimBoundaryThreshold { - return true - } - return endsWithSentencePunctuation(previous) -} - -// endsWithSentencePunctuation reports whether transcript looks sentence-complete. -func endsWithSentencePunctuation(transcript string) bool { - transcript = strings.TrimSpace(transcript) - if transcript == "" { - return false - } - switch transcript[len(transcript)-1] { - case '.', '!', '?': - return true - default: - return false - } -} - -// commonPrefixWords counts shared leading words across two slices. -func commonPrefixWords(left []string, right []string) int { - limit := len(left) - if len(right) < limit { - limit = len(right) - } - count := 0 - for i := 0; i < limit; i++ { - if left[i] != right[i] { - break - } - count++ - } - return count -} - -// commonSuffixWords counts shared trailing words across two slices. -func commonSuffixWords(left []string, right []string) int { - li := len(left) - 1 - ri := len(right) - 1 - count := 0 - for li >= 0 && ri >= 0 { - if left[li] != right[ri] { - break - } - count++ - li-- - ri-- - } - return count -} - // cleanSegment normalizes transcript whitespace. func cleanSegment(raw string) string { raw = strings.TrimSpace(raw) From d4df1d49a22b56fddaa17bbbfdec5df5e6b63c7a Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Mon, 23 Feb 2026 12:59:27 -0700 Subject: [PATCH 2/6] fix(riva): rebuild interim segment assembly for long dictation Restore segment-building behavior inside a single toggle session by sealing mature interim chains on divergence while still treating rewrite/correction updates as continuation. This preserves earlier speech when Riva drops or rewrites interim hypotheses during long utterances and avoids stale-prefix prepends. Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client.go | 13 ++-- apps/sotto/internal/riva/client_test.go | 73 ++++++++++++++--- apps/sotto/internal/riva/stream_receive.go | 14 +++- .../internal/riva/transcript_segments.go | 78 +++++++++++++++++++ 4 files changed, 161 insertions(+), 17 deletions(-) diff --git a/apps/sotto/internal/riva/client.go b/apps/sotto/internal/riva/client.go index 44f844a..a05a9b8 100644 --- a/apps/sotto/internal/riva/client.go +++ b/apps/sotto/internal/riva/client.go @@ -41,12 +41,13 @@ type Stream struct { recvDone chan struct{} - mu sync.Mutex - segments []string // committed transcript segments (final results only) - lastInterim string - recvErr error - closedSend bool - debugSinkJSON io.Writer + mu sync.Mutex + segments []string // committed transcript segments (final results and sealed interim chains) + lastInterim string + lastInterimAge int + recvErr error + closedSend bool + debugSinkJSON io.Writer } // DialStream establishes a stream, sends config, and starts the receive loop. diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index a9fc1be..595fd31 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,13 +82,12 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } -func TestRecordResponseDoesNotCommitDivergentInterimBeforeFinal(t *testing.T) { +func TestRecordResponseCommitsInterimChainOnDivergence(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.95, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase"}}, }}, }) @@ -86,14 +95,40 @@ func TestRecordResponseDoesNotCommitDivergentInterimBeforeFinal(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.20, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase extended"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "second phrase"}}, }}, }) - require.Empty(t, s.segments) + 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{"second phrase"}, segments) + 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) { @@ -102,7 +137,6 @@ func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, - Stability: 0.05, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "stale words"}}, }}, }) @@ -110,7 +144,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 +165,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 +172,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,9 +198,31 @@ func TestAppendSegmentDedupAndPrefixMerge(t *testing.T) { require.Equal(t, []string{"hello world", "new sentence"}, segments) } -func TestCleanSegment(t *testing.T) { +func TestInterimHelpers(t *testing.T) { require.Equal(t, "hello world", cleanSegment(" hello\n world ")) require.Empty(t, cleanSegment(" \n\t")) + + 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, shouldCommitInterimBoundary("", 5)) + require.False(t, shouldCommitInterimBoundary("first phrase", 1)) + require.True(t, shouldCommitInterimBoundary("first phrase", 2)) } func TestDialStreamEndToEndWithDebugSinkAndSpeechContexts(t *testing.T) { diff --git a/apps/sotto/internal/riva/stream_receive.go b/apps/sotto/internal/riva/stream_receive.go index 8c66834..0dac001 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -53,10 +53,22 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { if result.GetIsFinal() { s.segments = appendSegment(s.segments, transcript) s.lastInterim = "" + s.lastInterimAge = 0 continue } - // Keep only the latest interim hypothesis within a toggle session. + if s.lastInterim != "" { + if isInterimContinuation(s.lastInterim, transcript) { + s.lastInterim = transcript + s.lastInterimAge++ + continue + } + if shouldCommitInterimBoundary(s.lastInterim, s.lastInterimAge) { + s.segments = appendSegment(s.segments, s.lastInterim) + } + } + s.lastInterim = transcript + s.lastInterimAge = 1 } } diff --git a/apps/sotto/internal/riva/transcript_segments.go b/apps/sotto/internal/riva/transcript_segments.go index 2020752..e479061 100644 --- a/apps/sotto/internal/riva/transcript_segments.go +++ b/apps/sotto/internal/riva/transcript_segments.go @@ -2,6 +2,8 @@ package riva import "strings" +const minInterimChainUpdates = 2 + // collectSegments appends a valid trailing interim segment when needed. func collectSegments(committedSegments []string, lastInterim string) []string { segments := append([]string(nil), committedSegments...) @@ -35,6 +37,82 @@ func appendSegment(segments []string, transcript string) []string { } } +// 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) + if previous == "" || current == "" { + return true + } + if previous == current { + return true + } + if strings.HasPrefix(current, previous) || strings.HasPrefix(previous, current) { + return true + } + if strings.HasSuffix(current, previous) || strings.HasSuffix(previous, current) { + return true + } + + prevWords := strings.Fields(previous) + currWords := strings.Fields(current) + shorter := len(prevWords) + if len(currWords) < shorter { + shorter = len(currWords) + } + if shorter == 0 { + return true + } + + if commonPrefixWords(prevWords, currWords)*2 >= shorter { + return true + } + if shorter >= 3 && commonSuffixWords(prevWords, currWords)*2 >= shorter { + return true + } + + return false +} + +// shouldCommitInterimBoundary returns true when a divergent interim chain looks +// established enough to preserve as a committed segment. +func shouldCommitInterimBoundary(previous string, chainUpdates int) bool { + return cleanSegment(previous) != "" && chainUpdates >= minInterimChainUpdates +} + +// commonPrefixWords counts shared leading words across two slices. +func commonPrefixWords(left []string, right []string) int { + limit := len(left) + if len(right) < limit { + limit = len(right) + } + count := 0 + for i := 0; i < limit; i++ { + if left[i] != right[i] { + break + } + count++ + } + return count +} + +// commonSuffixWords counts shared trailing words across two slices. +func commonSuffixWords(left []string, right []string) int { + li := len(left) - 1 + ri := len(right) - 1 + count := 0 + for li >= 0 && ri >= 0 { + if left[li] != right[ri] { + break + } + count++ + li-- + ri-- + } + return count +} + // cleanSegment normalizes transcript whitespace. func cleanSegment(raw string) string { raw = strings.TrimSpace(raw) From a79449378d3f3fd2c4439270ee67b1adb26c2aae Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Mon, 23 Feb 2026 12:59:51 -0700 Subject: [PATCH 3/6] refactor(pipeline): add injectable seams for transcriber tests Introduce capture/stream interfaces and dependency hooks in pipeline.Transcriber so lifecycle logic can be tested without live Pulse/Riva runtime dependencies. Also adds broad transcriber unit coverage for start/stop/cancel/send-loop success and error paths. Tests: - go test ./apps/sotto/internal/pipeline - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/pipeline/transcriber.go | 43 ++- .../internal/pipeline/transcriber_test.go | 257 ++++++++++++++++++ 2 files changed, 294 insertions(+), 6 deletions(-) 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 +} From d6b51096ef64682c82660ea3057febaa003af5dd Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Mon, 23 Feb 2026 13:00:08 -0700 Subject: [PATCH 4/6] test(internal): expand coverage across app audio config doctor Add focused unit tests across remaining lower-coverage internal modules: - app: status/socket helpers, toggle owner startup failure cleanup, session log fields - audio: pulse-unavailable selection/listing paths and capture close/device wrappers - config: JSONC normalization/decoder/field-validation edge cases - doctor: report + runtime check routing for paste override vs hyprctl defaults Tests: - go test ./apps/sotto/internal/... -cover - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/app/app_test.go | 88 +++++++++++ apps/sotto/internal/audio/pulse_test.go | 26 ++++ .../internal/config/parser_jsonc_test.go | 147 ++++++++++++++++++ apps/sotto/internal/doctor/doctor_test.go | 62 ++++++++ 4 files changed, 323 insertions(+) create mode 100644 apps/sotto/internal/config/parser_jsonc_test.go 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_test.go b/apps/sotto/internal/config/parser_jsonc_test.go new file mode 100644 index 0000000..a8aafef --- /dev/null +++ b/apps/sotto/internal/config/parser_jsonc_test.go @@ -0,0 +1,147 @@ +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.True( + t, + strings.Contains(err.Error(), "multiple JSON values") || strings.Contains(err.Error(), "unknown field"), + "unexpected error: %v", + err, + ) +} + +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) +} From 120312e335aa3fab7bfb241a095908b71593e3d2 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Mon, 23 Feb 2026 13:44:04 -0700 Subject: [PATCH 5/6] fix(config): enforce single top-level JSONC value Make multi-value JSON payloads fail with a stable "multiple JSON values" error and tighten parser regression coverage accordingly. Tests: - go test ./apps/sotto/internal/config - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/config/parser_jsonc.go | 2 +- apps/sotto/internal/config/parser_jsonc_test.go | 7 +------ 2 files changed, 2 insertions(+), 7 deletions(-) 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 index a8aafef..add8328 100644 --- a/apps/sotto/internal/config/parser_jsonc_test.go +++ b/apps/sotto/internal/config/parser_jsonc_test.go @@ -114,12 +114,7 @@ func TestParseJSONCTrimsIndicatorAndPasteFields(t *testing.T) { func TestParseJSONCRejectsMultipleTopLevelValues(t *testing.T) { _, _, err := parseJSONC(`{"paste":{"enable":false}}{"paste":{"enable":true}}`, Default()) require.Error(t, err) - require.True( - t, - strings.Contains(err.Error(), "multiple JSON values") || strings.Contains(err.Error(), "unknown field"), - "unexpected error: %v", - err, - ) + require.Contains(t, err.Error(), "multiple JSON values") } func TestParseJSONCTypeErrorIncludesLocation(t *testing.T) { From 63ca29e261def039327d850e891054d2b9129aa0 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Mon, 23 Feb 2026 13:47:11 -0700 Subject: [PATCH 6/6] fix(riva): keep one-shot interim segments when boundary is reliable Address PR feedback on divergence handling by preserving single-update interim segments when they appear reliable (high stability or sentence punctuation), while retaining age-based chain commits and continuation matching. Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client.go | 15 +++++---- apps/sotto/internal/riva/client_test.go | 32 +++++++++++++++++-- apps/sotto/internal/riva/stream_receive.go | 5 ++- .../internal/riva/transcript_segments.go | 32 +++++++++++++++++-- 4 files changed, 70 insertions(+), 14 deletions(-) diff --git a/apps/sotto/internal/riva/client.go b/apps/sotto/internal/riva/client.go index a05a9b8..6274366 100644 --- a/apps/sotto/internal/riva/client.go +++ b/apps/sotto/internal/riva/client.go @@ -41,13 +41,14 @@ type Stream struct { recvDone chan struct{} - mu sync.Mutex - segments []string // committed transcript segments (final results and sealed interim chains) - lastInterim string - lastInterimAge int - recvErr error - closedSend bool - debugSinkJSON io.Writer + mu sync.Mutex + segments []string // committed transcript segments (final results and sealed interim chains) + lastInterim string + lastInterimAge int + lastInterimStability float32 + recvErr error + closedSend bool + debugSinkJSON io.Writer } // DialStream establishes a stream, sends config, and starts the receive loop. diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index 595fd31..cd0e9a4 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -82,6 +82,30 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } +func TestRecordResponseCommitsStableSingleInterimOnDivergence(t *testing.T) { + s := &Stream{} + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Stability: 0.95, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Stability: 0.20, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "second phrase"}}, + }}, + }) + + require.Equal(t, []string{"first phrase"}, s.segments) + segments := collectSegments(s.segments, s.lastInterim) + require.Equal(t, []string{"first phrase", "second phrase"}, segments) +} + func TestRecordResponseCommitsInterimChainOnDivergence(t *testing.T) { s := &Stream{} @@ -220,9 +244,11 @@ func TestInterimHelpers(t *testing.T) { }) } - require.False(t, shouldCommitInterimBoundary("", 5)) - require.False(t, shouldCommitInterimBoundary("first phrase", 1)) - require.True(t, shouldCommitInterimBoundary("first phrase", 2)) + 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 0dac001..e7801fe 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -54,6 +54,7 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { s.segments = appendSegment(s.segments, transcript) s.lastInterim = "" s.lastInterimAge = 0 + s.lastInterimStability = 0 continue } @@ -61,14 +62,16 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { if isInterimContinuation(s.lastInterim, transcript) { s.lastInterim = transcript s.lastInterimAge++ + s.lastInterimStability = result.GetStability() continue } - if shouldCommitInterimBoundary(s.lastInterim, s.lastInterimAge) { + 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 e479061..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 minInterimChainUpdates = 2 +const ( + minInterimChainUpdates = 2 + stableInterimBoundaryThreshold = 0.85 +) // collectSegments appends a valid trailing interim segment when needed. func collectSegments(committedSegments []string, lastInterim string) []string { @@ -77,8 +80,31 @@ func isInterimContinuation(previous string, 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) bool { - return cleanSegment(previous) != "" && chainUpdates >= minInterimChainUpdates +func shouldCommitInterimBoundary(previous string, chainUpdates int, stability float32) bool { + previous = cleanSegment(previous) + if previous == "" { + return false + } + if chainUpdates >= minInterimChainUpdates { + return true + } + if stability >= stableInterimBoundaryThreshold { + return true + } + return endsWithSentencePunctuation(previous) +} + +func endsWithSentencePunctuation(text string) bool { + text = strings.TrimSpace(text) + if text == "" { + return false + } + switch text[len(text)-1] { + case '.', '!', '?': + return true + default: + return false + } } // commonPrefixWords counts shared leading words across two slices.