diff --git a/docs/tentacle-remote-speech.md b/docs/tentacle-remote-speech.md index 4a3456e..c6b6d0d 100644 --- a/docs/tentacle-remote-speech.md +++ b/docs/tentacle-remote-speech.md @@ -359,10 +359,10 @@ track a backend speech position. It does not change Speech Dispatcher priority or establish scheduling parity with NVDA. Cancel discards pending speech and cancels only the active remote renderer. -Pause true does the same and discards incoming speech while paused. Pause false -accepts future speech; it never resumes discarded text. Cancel does not unpause -the stream. A replacement lease resets pause state, discards old pending work, -and waits for old renderer cleanup before starting new output. Stale/cancelled +Pause true does the same; it does not mute future utterances. NVDA clears its +pause state on cancellation or fresh speech without requiring a separate resume +message. Pause false never replays discarded text. A replacement lease discards +old pending work and waits for old renderer cleanup before starting new output. Stale/cancelled deliveries are released without rendering. Lease cancellation and owner EOF or shutdown also cancel active work and release retained pending budget. diff --git a/internal/app/remote_speech_stream.go b/internal/app/remote_speech_stream.go index a23f567..6319ce5 100644 --- a/internal/app/remote_speech_stream.go +++ b/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) diff --git a/internal/app/remote_speech_stream_test.go b/internal/app/remote_speech_stream_test.go index 602f9ba..dc54dcd 100644 --- a/internal/app/remote_speech_stream_test.go +++ b/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{}) diff --git a/internal/speech/read_all.go b/internal/speech/read_all.go index 3faa6ff..738db1d 100644 --- a/internal/speech/read_all.go +++ b/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) } diff --git a/internal/speech/read_all_test.go b/internal/speech/read_all_test.go index 686745c..5c87906 100644 --- a/internal/speech/read_all_test.go +++ b/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") + } + }) + } +} diff --git a/internal/speech/speech.go b/internal/speech/speech.go index c4de909..d98d3d9 100644 --- a/internal/speech/speech.go +++ b/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) + } } }