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'

Author: rbright

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.

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) {

apps/sotto/internal/riva/stream_receive.go

@@ -54,21 +54,24 @@ func (s *Stream) recordResponse(resp *asrpb.StreamingRecognizeResponse) {
 			s.segments = appendSegment(s.segments, transcript)
 			s.lastInterim = ""
 			s.lastInterimAge = 0
+			s.lastInterimStability = 0
 			continue
 		}
 
 		if s.lastInterim != "" {
 			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()
 	}
 }

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.