From 38d27985055b242e7b38297f35defd2169b47a41 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Sun, 22 Feb 2026 17:17:18 -0700 Subject: [PATCH 1/4] fix(riva): avoid stale interim carryover in segment assembly Do not pre-commit divergent interim hypotheses before final results. Riva can reset interim boundaries between updates; pre-committing the prior interim can prepend stale text or duplicate leading phrases in the final transcript. Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client_test.go | 33 ++++++++++++++++++++-- apps/sotto/internal/riva/stream_receive.go | 6 ++-- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index 679103f..c572165 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -50,7 +50,7 @@ func TestRecordResponseTracksInterimThenFinal(t *testing.T) { require.Equal(t, []string{"hello world"}, s.segments) } -func TestRecordResponseCommitsInterimAcrossPauseLikeReset(t *testing.T) { +func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ @@ -67,8 +67,37 @@ func TestRecordResponseCommitsInterimAcrossPauseLikeReset(t *testing.T) { }}, }) + 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) { + s := &Stream{} + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "stale words"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "hello world"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: true, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "hello world"}}, + }}, + }) + + segments := collectSegments(s.segments, s.lastInterim) + require.Equal(t, []string{"hello world"}, segments) } func TestAppendSegmentDedupAndPrefixMerge(t *testing.T) { diff --git a/apps/sotto/internal/riva/stream_receive.go b/apps/sotto/internal/riva/stream_receive.go index ef0e226..6028e20 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -56,9 +56,9 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { continue } - if s.lastInterim != "" && !isInterimContinuation(s.lastInterim, transcript) { - s.segments = appendSegment(s.segments, s.lastInterim) - } + // Keep only the latest interim hypothesis. Riva can reset interim text + // boundaries between updates; pre-committing the prior interim here can + // introduce duplicated or stale leading segments in the final transcript. s.lastInterim = transcript } } From 52817f008886a6507bcdf5ee33487a78ceb361f5 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Sun, 22 Feb 2026 18:20:06 -0700 Subject: [PATCH 2/4] fix(riva): preserve stable interim segments on divergence Address PR review feedback by preserving prior interim text only when divergence indicates a likely utterance boundary. - track interim stability in stream state - commit prior interim on divergence only when high-confidence or sentence-complete - keep low-confidence divergence replacement to avoid stale/duplicated leading text - add regression coverage for stable partial recovery and stale-prepend avoidance 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 | 30 +++++++++++++++++ apps/sotto/internal/riva/stream_receive.go | 8 +++-- .../internal/riva/transcript_segments.go | 33 +++++++++++++++++++ 4 files changed, 75 insertions(+), 9 deletions(-) diff --git a/apps/sotto/internal/riva/client.go b/apps/sotto/internal/riva/client.go index dee7031..bcc272b 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 and pause-committed interim) - lastInterim string - recvErr error - closedSend bool - debugSinkJSON io.Writer + 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 } // 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 c572165..07210f1 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -72,12 +72,37 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } +func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(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 TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, + Stability: 0.05, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "stale words"}}, }}, }) @@ -85,6 +110,7 @@ func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, + Stability: 0.30, Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "hello world"}}, }}, }) @@ -124,6 +150,10 @@ func TestCleanSegmentAndInterimContinuation(t *testing.T) { require.True(t, isInterimContinuation("hello", "hello world")) require.True(t, isInterimContinuation("hello world", "hello")) 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, "new sentence")) } func TestDialStreamEndToEndWithDebugSinkAndSpeechContexts(t *testing.T) { diff --git a/apps/sotto/internal/riva/stream_receive.go b/apps/sotto/internal/riva/stream_receive.go index 6028e20..769e642 100644 --- a/apps/sotto/internal/riva/stream_receive.go +++ b/apps/sotto/internal/riva/stream_receive.go @@ -53,12 +53,14 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) { if result.GetIsFinal() { s.segments = appendSegment(s.segments, transcript) s.lastInterim = "" + s.lastInterimStability = 0 continue } - // Keep only the latest interim hypothesis. Riva can reset interim text - // boundaries between updates; pre-committing the prior interim here can - // introduce duplicated or stale leading segments in the final transcript. + if shouldCommitPriorInterimOnDivergence(s.lastInterim, s.lastInterimStability, transcript) { + s.segments = appendSegment(s.segments, s.lastInterim) + } 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 0e887f6..97d84d5 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 stableInterimBoundaryThreshold = 0.85 + // collectSegments appends a valid trailing interim segment when needed. func collectSegments(committedSegments []string, lastInterim string) []string { segments := append([]string(nil), committedSegments...) @@ -62,6 +64,37 @@ func isInterimContinuation(previous string, current string) bool { return common*2 >= shorter } +// 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) From d88888035bc3d50171bdf5fd939a8d6a6213e051 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Sun, 22 Feb 2026 18:47:24 -0700 Subject: [PATCH 3/4] fix(riva): tighten interim boundary commit guard Only preserve divergent interim text when it looks like a completed sentence and has high stability. This avoids prepending corrected-but-stale leading words while still keeping partial-recovery support for likely utterance boundaries in no-final EOF paths. Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client_test.go | 13 +++++++------ apps/sotto/internal/riva/transcript_segments.go | 4 ++-- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index 07210f1..420d77e 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -72,14 +72,14 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } -func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testing.T) { +func TestRecordResponseCommitsStableSentenceInterimForPartialRecovery(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, Stability: 0.95, - Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase"}}, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase."}}, }}, }) @@ -91,9 +91,9 @@ func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testin }}, }) - require.Equal(t, []string{"first phrase"}, s.segments) + require.Equal(t, []string{"first phrase."}, s.segments) segments := collectSegments(s.segments, s.lastInterim) - require.Equal(t, []string{"first phrase", "second phrase"}, segments) + require.Equal(t, []string{"first phrase.", "second phrase"}, segments) } func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { @@ -152,8 +152,9 @@ func TestCleanSegmentAndInterimContinuation(t *testing.T) { 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, "new sentence")) + require.False(t, shouldCommitPriorInterimOnDivergence("first phrase", 0.9, "second phrase")) + require.True(t, shouldCommitPriorInterimOnDivergence("Done.", 0.9, "new sentence")) + require.False(t, shouldCommitPriorInterimOnDivergence("Done.", 0.1, "new sentence")) } func TestDialStreamEndToEndWithDebugSinkAndSpeechContexts(t *testing.T) { diff --git a/apps/sotto/internal/riva/transcript_segments.go b/apps/sotto/internal/riva/transcript_segments.go index 97d84d5..08c122e 100644 --- a/apps/sotto/internal/riva/transcript_segments.go +++ b/apps/sotto/internal/riva/transcript_segments.go @@ -75,8 +75,8 @@ func shouldCommitPriorInterimOnDivergence(previous string, previousStability flo if isInterimContinuation(previous, current) { return false } - if previousStability >= stableInterimBoundaryThreshold { - return true + if previousStability < stableInterimBoundaryThreshold { + return false } return endsWithSentencePunctuation(previous) } From f2f7d89740fba957705f6d5b53104692dfaa9f58 Mon Sep 17 00:00:00 2001 From: Ryan Bright Date: Sun, 22 Feb 2026 19:11:02 -0700 Subject: [PATCH 4/4] fix(riva): retain long-stream context without stale prefix commit Relax divergent interim boundary commits to preserve stable partial speech in long dictation while treating suffix-based corrections as continuation updates. - commit divergent interim when prior stability is high or sentence punctuation is present - detect suffix-overlap corrections and avoid pre-committing stale leading words - add regression tests for long-stream partial recovery and suffix correction behavior Tests: - go test ./apps/sotto/internal/riva - just ci-check - nix build 'path:.#sotto' --- apps/sotto/internal/riva/client_test.go | 39 +++++++++++++++---- .../internal/riva/transcript_segments.go | 37 ++++++++++++++++-- 2 files changed, 65 insertions(+), 11 deletions(-) diff --git a/apps/sotto/internal/riva/client_test.go b/apps/sotto/internal/riva/client_test.go index 420d77e..c7afc9e 100644 --- a/apps/sotto/internal/riva/client_test.go +++ b/apps/sotto/internal/riva/client_test.go @@ -72,14 +72,14 @@ func TestRecordResponseReplacesDivergentInterimWithoutPrecommit(t *testing.T) { require.Equal(t, []string{"second phrase"}, segments) } -func TestRecordResponseCommitsStableSentenceInterimForPartialRecovery(t *testing.T) { +func TestRecordResponseCommitsStableDivergentInterimForPartialRecovery(t *testing.T) { s := &Stream{} s.recordResponse(&asrpb.StreamingRecognizeResponse{ Results: []*asrpb.StreamingRecognitionResult{{ IsFinal: false, Stability: 0.95, - Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase."}}, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "first phrase"}}, }}, }) @@ -91,9 +91,9 @@ func TestRecordResponseCommitsStableSentenceInterimForPartialRecovery(t *testing }}, }) - require.Equal(t, []string{"first phrase."}, s.segments) + require.Equal(t, []string{"first phrase"}, s.segments) segments := collectSegments(s.segments, s.lastInterim) - require.Equal(t, []string{"first phrase.", "second phrase"}, segments) + require.Equal(t, []string{"first phrase", "second phrase"}, segments) } func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { @@ -126,6 +126,30 @@ func TestRecordResponseDoesNotPrependStaleInterimBeforeFinal(t *testing.T) { require.Equal(t, []string{"hello world"}, segments) } +func TestRecordResponseTreatsSuffixCorrectionAsContinuation(t *testing.T) { + s := &Stream{} + + 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"}}, + }}, + }) + + s.recordResponse(&asrpb.StreamingRecognizeResponse{ + Results: []*asrpb.StreamingRecognitionResult{{ + IsFinal: false, + Stability: 0.95, + Alternatives: []*asrpb.SpeechRecognitionAlternative{{Transcript: "replied on the review thread with details"}}, + }}, + }) + + require.Empty(t, s.segments) + segments := collectSegments(s.segments, s.lastInterim) + require.Equal(t, []string{"replied on the review thread with details"}, segments) +} + func TestAppendSegmentDedupAndPrefixMerge(t *testing.T) { segments := appendSegment(nil, "hello") require.Equal(t, []string{"hello"}, segments) @@ -149,12 +173,13 @@ func TestCleanSegmentAndInterimContinuation(t *testing.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.False(t, shouldCommitPriorInterimOnDivergence("first phrase", 0.9, "second phrase")) - require.True(t, shouldCommitPriorInterimOnDivergence("Done.", 0.9, "new sentence")) - require.False(t, shouldCommitPriorInterimOnDivergence("Done.", 0.1, "new sentence")) + 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/transcript_segments.go b/apps/sotto/internal/riva/transcript_segments.go index 08c122e..ea7efb2 100644 --- a/apps/sotto/internal/riva/transcript_segments.go +++ b/apps/sotto/internal/riva/transcript_segments.go @@ -50,10 +50,12 @@ func isInterimContinuation(previous string, current string) bool { 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) - common := commonPrefixWords(prevWords, currWords) shorter := len(prevWords) if len(currWords) < shorter { shorter = len(currWords) @@ -61,7 +63,18 @@ func isInterimContinuation(previous string, current string) bool { if shorter == 0 { return true } - return common*2 >= shorter + + 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 @@ -75,8 +88,8 @@ func shouldCommitPriorInterimOnDivergence(previous string, previousStability flo if isInterimContinuation(previous, current) { return false } - if previousStability < stableInterimBoundaryThreshold { - return false + if previousStability >= stableInterimBoundaryThreshold { + return true } return endsWithSentencePunctuation(previous) } @@ -111,6 +124,22 @@ func commonPrefixWords(left []string, right []string) int { 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)