Fix Remote speech stalls after pause and early completion

internal/app/remote_speech_stream.go

@@ -111,7 +111,6 @@ func run_remote_speech_stream(ctx context.Context, incoming <-chan tentacle.Remo
 	var activeContext context.Context
 	var lease tentaclewire.RemoteLease
 	var leaseDone <-chan struct{}
-	paused := false
 	items, bytes := 0, 0
 	valid := func(delivery tentacle.RemoteSpeechDelivery) bool {
 		return ctx.Err() == nil && delivery.Context != nil && delivery.Context.Err() == nil && allow(delivery)
@@ -187,7 +186,7 @@ func run_remote_speech_stream(ctx context.Context, incoming <-chan tentacle.Remo
 			}
 			if lease != delivery.Lease {
 				interrupt()
-				lease, leaseDone, paused = delivery.Lease, delivery.Context.Done(), false
+				lease, leaseDone = delivery.Lease, delivery.Context.Done()
 			}
 			if len(delivery.Speeches) == 1 {
 				effect := delivery.Speeches[0]
@@ -195,9 +194,10 @@ func run_remote_speech_stream(ctx context.Context, incoming <-chan tentacle.Remo
 					if effect.Kind == "cancel" || *effect.Paused {
 						interrupt()
 					}
-					if effect.Kind == "pause" {
-						paused = *effect.Paused
-					}
+					// Pause cancels the current assembly because this backend
+					// cannot resume it. It must not mute future utterances: NVDA
+					// clears pause on cancel or fresh speech, without requiring
+					// a separate pause(false) message on the relay.
 					delivery.Release()
 					continue
 				}
@@ -207,10 +207,6 @@ func run_remote_speech_stream(ctx context.Context, incoming <-chan tentacle.Remo
 				delivery.Release()
 				return &remoteSpeechStreamError{lease: delivery.Lease, cause: err}
 			}
-			if paused {
-				delivery.Release()
-				continue
-			}
 			permit := remoteSpeechPermit{context: delivery.Context, release: func() {}, current: func() bool { return true }}
 			if admit != nil {
 				admitted, ok := admit(delivery)

internal/app/remote_speech_stream_test.go

@@ -7,6 +7,7 @@ package app
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -188,10 +189,6 @@ func TestRemoteSpeechStreamCancelAndPause(t *testing.T) {
 			remote_stream_wait(t, f.send(t, effect))
 			remote_stream_wait(t, active.stopped)
 			remote_stream_wait(t, pendingRelease)
-			if kind == "pause" {
-				remote_stream_wait(t, f.speak(t, "discard while paused", 2))
-				f.barrier(t) // Resume accepts future speech only.
-			}
 			futureRelease := f.speak(t, "future", 0)
 			close(active.cleanup)
 			future := f.next(t, "future")
@@ -203,6 +200,24 @@ func TestRemoteSpeechStreamCancelAndPause(t *testing.T) {
 	}
 }
 
+func TestRemoteSpeechStreamFreshSpeechAfterPause(t *testing.T) {
+	for _, cancelFirst := range []bool{false, true} {
+		t.Run(map[bool]string{false: "speak", true: "cancel then speak"}[cancelFirst], func(t *testing.T) {
+			f := remote_stream_fixture(t, false)
+			paused := true
+			remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "pause", Paused: &paused}))
+			if cancelFirst {
+				remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "cancel"}))
+			}
+			released := f.speak(t, "new navigation", 0)
+			call := f.next(t, "new navigation")
+			call.finish <- nil
+			remote_stream_wait(t, released)
+			f.stop(t)
+		})
+	}
+}
+
 func TestRemoteSpeechStreamRevocationAndReplacement(t *testing.T) {
 	f := remote_stream_fixture(t, true)
 	leaseContext, revoke := context.WithCancel(f.ctx)
@@ -349,6 +364,40 @@ func TestRemoteSpeechStreamIsolatedBackendCancellation(t *testing.T) {
 	}
 }
 
+func TestRemoteSpeechStreamEarlyBackendCompletionKeepsReading(t *testing.T) {
+	for _, terminal := range []int{702, 703} {
+		t.Run(fmt.Sprint(terminal), func(t *testing.T) {
+			speaker, server := isolated_voice_fixture(t)
+			server.mutex.Lock()
+			server.queryReplies = map[string]string{"SPEAK": fmt.Sprintf(
+				"230 OK\r\n%d-1\r\n%d-1\r\n%d DONE\r\n225-1\r\n225 QUEUED\r\n", terminal, terminal, terminal)}
+			server.mutex.Unlock()
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			incoming := make(chan tentacle.RemoteSpeechDelivery, 8)
+			done := make(chan error, 1)
+			go func() {
+				done <- run_remote_speech_playback(ctx, speaker, incoming, func(tentacle.RemoteSpeechDelivery) bool { return true }, nil)
+			}()
+			lease := tentaclewire.RemoteLease{Lease: strings.Repeat("b", 32), Epoch: 1}
+			for _, text := range []string{"first", "second", "third"} {
+				released := make(chan struct{})
+				incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease,
+					Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: text})},
+					Release:  func() { close(released) }}
+				if got := server.next_utterance(t); got.text != text {
+					t.Fatalf("remote output = %q, want %q", got.text, text)
+				}
+				remote_stream_wait(t, released)
+			}
+			cancel()
+			if err := remote_render_result(t, done); !errors.Is(err, context.Canceled) {
+				t.Fatalf("stream shutdown: %v", err)
+			}
+		})
+	}
+}
+
 func TestRemoteSpeechStreamPolicyLossDuringAcquisition(t *testing.T) {
 	speaker, server := isolated_voice_fixture(t)
 	held, resume := make(chan struct{}), make(chan struct{})

internal/speech/read_all.go

@@ -38,6 +38,29 @@ func (client *Client) speak_and_wait_locked(ctx context.Context, speak func() er
 			}
 		}
 	}()
+	// Notifications can precede the SPEAK queue acknowledgement. Handle them
+	// while reading command replies too, or an early END/CANCEL is lost and
+	// this owner waits forever while later remote utterances queue behind it.
+	started, completed := false, false
+	var completion error
+	handle := func(event reply) {
+		if completed {
+			return
+		}
+		switch event.code {
+		case 701:
+			if !started && ctx.Err() == nil && began != nil {
+				began()
+			}
+			started = true
+		case 702:
+			completed, completion = true, ctx.Err()
+		case 703:
+			completed, completion = true, context.Canceled
+		}
+	}
+	client.notification = handle
+	defer func() { client.notification = nil }()
 	if err := speak(); err != nil {
 		return err
 	}
@@ -61,24 +84,14 @@ func (client *Client) speak_and_wait_locked(ctx context.Context, speak func() er
 			<-done
 		}
 	}()
-	started := false
-	for {
+	for !completed {
 		event, err := readReply(client.reader)
 		if err != nil {
 			_ = conn.Close()
 			client.conn, client.reader = nil, nil
 			return errors.Join(ctx.Err(), err)
 		}
-		switch event.code {
-		case 701:
-			if !started && ctx.Err() == nil && began != nil {
-				began()
-			}
-			started = true
-		case 702:
-			return ctx.Err()
-		case 703:
-			return context.Canceled
-		}
+		handle(event)
 	}
+	return errors.Join(ctx.Err(), completion)
 }

internal/speech/read_all_test.go

@@ -82,3 +82,42 @@ func TestContinuousSpeechWaitsForEndAndSurvivesCancellation(t *testing.T) {
 		})
 	}
 }
+
+func TestContinuousSpeechCompletionBeforeQueueReply(t *testing.T) {
+	for _, terminal := range []int{702, 703} {
+		t.Run(fmt.Sprint(terminal), func(t *testing.T) {
+			conn, server := net.Pipe()
+			client := connectedClient(conn)
+			t.Cleanup(func() { server.Close(); client.Close(context.Background()) })
+			began := make(chan struct{}, 1)
+			finished := make(chan error, 1)
+			go func() {
+				finished <- client.SpeakAndWait(t.Context(), "Short remote response", func() { began <- struct{}{} })
+			}()
+			server.SetDeadline(time.Now().Add(2 * time.Second))
+			reader := bufio.NewReader(server)
+			if err := exchangeSSIP(reader, server, "SET SELF NOTIFICATION ALL on\r\n", "200 OK\r\n"); err != nil {
+				t.Fatal(err)
+			}
+			for range 3 {
+				if _, err := reader.ReadString('\n'); err != nil {
+					t.Fatal(err)
+				}
+			}
+			// Speech Dispatcher queues work before returning the SPEAK reply.
+			// A short or preempted utterance can finish before that reply arrives.
+			fmt.Fprintf(server, "230 OK\r\n701-1\r\n701-42\r\n701 BEGIN\r\n%d-1\r\n%d-42\r\n%d DONE\r\n225-1\r\n225 QUEUED\r\n", terminal, terminal, terminal)
+			if err := exchangeSSIP(reader, server, "SET SELF NOTIFICATION ALL off\r\n", "200 OK\r\n"); err != nil {
+				t.Fatal("completion before queue reply was lost:", err)
+			}
+			if err := <-finished; terminal == 703 && !errors.Is(err, context.Canceled) || terminal == 702 && err != nil {
+				t.Fatalf("completion = %v", err)
+			}
+			select {
+			case <-began:
+			default:
+				t.Fatal("early BEGIN was lost")
+			}
+		})
+	}
+}

internal/speech/speech.go

@@ -47,6 +47,7 @@ type Client struct {
 	mutex         sync.Mutex
 	cancelMutex   sync.Mutex
 	settingsMutex sync.Mutex
+	notification  func(reply) // Speaking mutex owns the active completion handler.
 }
 
 type Settings struct {
@@ -426,6 +427,9 @@ func (client *Client) readReply() (reply, error) {
 		if err != nil || response.code < 700 || response.code >= 800 {
 			return response, err
 		}
+		if client.notification != nil {
+			client.notification(response)
+		}
 	}
 }