From 914d1749256f29e612b257afc5ce11c92b8adcfc Mon Sep 17 00:00:00 2001 From: Angus Kola Date: Wed, 4 Nov 2026 20:10:47 -0500 Subject: [PATCH] Add bounded incoming Remote speech stream ownership --- docs/tentacle-remote-speech.md | 77 +++- internal/app/remote_speech_stream.go | 218 +++++++++++ internal/app/remote_speech_stream_test.go | 447 ++++++++++++++++++++++ 3 files changed, 733 insertions(+), 9 deletions(-) create mode 100644 internal/app/remote_speech_stream.go create mode 100644 internal/app/remote_speech_stream_test.go diff --git a/docs/tentacle-remote-speech.md b/docs/tentacle-remote-speech.md index b2ab999..4e13779 100644 --- a/docs/tentacle-remote-speech.md +++ b/docs/tentacle-remote-speech.md @@ -225,11 +225,59 @@ contract](https://github.com/brailcom/speechd/blob/master/doc/ssip.texi) and che default-query/module-switch behavior against the upstream [parser](https://github.com/brailcom/speechd/blob/master/src/server/parse.c) and [settings implementation](https://github.com/brailcom/speechd/blob/master/src/server/set.c). -It makes no NVDA priority or pause decision. A production owner still has to -apply priority once, recheck original-lease/focus/mode policy, yield to local -output, process cancel/pause while synthesis is outstanding, and retain mailbox -budget until completion/discard. The source Remote command remains relay-only -for incoming speech; no active or installed reader was replaced. +The renderer itself makes no priority or pause decision. The stream prerequisite +below now applies those decisions and retains mailbox budget, but a production +owner still needs original-lease/focus/mode policy and arbitration with local +output. The source Remote command remains relay-only for incoming speech; +no active or installed reader was replaced. + +### Incoming stream prerequisite + +`run_remote_speech_playback` now joins the isolated renderer to an explicit +delivery channel. It is not registered with production sessions and provides +no default authorization callback. Its caller must supply the reader's +original-lease/focus/output check, bind policy loss to the delivery lifetime, +and retire the session on owner failure. Policy is checked at admission, before +dispatch and after backend acquisition. These callbacks are not a substitute +for production local-output arbitration, which remains pending. + +One owner receives complete assemblies and control effects while a separate +renderer performs backend I/O. It retains the original delivery's `Release` +until completion or discard, including through cancelled-renderer cleanup. +Pending work is bounded independently to the existing 64 frames and 256 KiB, +including the active assembly. Invalid assemblies, exceeded budgets and backend +failure end the owner; errors omit speech and backend diagnostics. The future +session caller must revoke/drain remaining ingress on failure or shutdown. + +Normal speech is FIFO. NEXT moves ahead of pending normal assemblies but waits +for the active assembly, including its internal end boundaries. NOW interrupts +an active lower-priority assembly and precedes pending lower-priority work; +equal priorities remain FIFO. Interrupted speech is discarded, without replay +or suffix retry. This is deliberately limited priority support: NVDA can resume +interrupted speech and preempt at utterance boundaries; this renderer does not +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 +deliveries are released without rendering. Lease cancellation and owner EOF or +shutdown also cancel active work and release retained pending budget. + +The software handle is reused within its lease after successful rendering. A +cancelled rendering retires it; the next accepted assembly acquires a fresh +handle after cleanup. Even an idle handle closes asynchronously on cancellation +of its original lease. Replacement waits for that closure and acquires its own +client; repeated old cleanup cannot cancel the replacement or local status. +Remote synthesis remains outside local history and observation paths. + +This prerequisite changes no Xorg capture, keyboard focus, installed reader or +production session routing. Private SSIP fixtures establish scoped protocol +ownership and cancellation, without audible speech or real server scheduling. +Production mailbox attachment, local output priority, physical input and live +NVDA/Orca acceptance remain pending. ### Automated evidence @@ -252,6 +300,17 @@ existing speech regression harness, affected-package vet, separate reader build and Tentacles `make check` passed. These are protocol fixtures, without real Speech Dispatcher, audio, physical input or Braille acceptance. +The incoming-stream increment passed ten repetitions of its focused race cases +and the existing application Remote speech and echo race +checks. Deterministic fixtures cover NEXT/NOW ordering, equal-priority FIFO, +cancel/pause while rendering, no replay, queued revocation, replacement leases, +policy rejection, frame/byte overflow, backend failure and EOF cleanup. Private +SSIP checks retain local status across stream cancellation, reject policy loss +during acquisition, close idle voices on revoke and protect replacement voices +from old cleanup. The existing speech regression harness, application vet, +separate reader build and Tentacles `make check` passed. No live acceptance +or full reader suite was performed for this unwired increment. + Run from this checkout: ```sh @@ -286,14 +345,14 @@ host/wire/both-SDK race suites (host: about 127 seconds), focused application speech/Remote/Tentacle race checks, affected-package vet, a separate reader build, the existing speech regression harness and Tentacles `make check`. -Next: bind the isolated renderer to incoming speech ownership, delivery-time -policy, local priority and stream cancellation. Explicit status detachment is implemented; physical +Next: attach the stream owner to production sessions with delivery-time +policy and local output priority. Explicit status detachment is implemented; physical input capture/release and ordinary reading while connected still need acceptance. Review the remaining output producers before assigning exportable origins. Atomic IPC assembly is implemented; it does not establish backend priority/cancel behavior or audible delivery. -Pause support or cancel-on-pause behavior and NVDA priority fidelity must be -resolved there. The codec transports metadata without claiming backend behavior. +The stream prerequisite implements cancel-on-pause and limited priority as +documented above; full NVDA priority/resumption fidelity is not established. Then verify normal reading, key echo, read-all, spelling, speech modes and local status while connected; lock/handoff and disconnect must discard remote work. Physical consent, audible speech and real NVDA/Orca acceptance remain pending. diff --git a/internal/app/remote_speech_stream.go b/internal/app/remote_speech_stream.go new file mode 100644 index 0000000..32651be --- /dev/null +++ b/internal/app/remote_speech_stream.go @@ -0,0 +1,218 @@ +// Copyright (c) 2026 Storm Dragon +// Authors: Storm Dragon, Angus Kola +// SPDX-License-Identifier: GPL-3.0-or-later + +package app + +import ( + "context" + "encoding/json" + "errors" + "time" + + "git.stormux.org/storm/cthulhu/internal/tentacle" + "git.stormux.org/storm/cthulhu/internal/tentaclewire" +) + +type remoteSpeechStreamItem struct { + delivery tentacle.RemoteSpeechDelivery + utterance remoteSpeechUtterance + bytes int +} + +// run_remote_speech_playback supplies the isolated backend to the stream owner. +// It is deliberately not registered with production sessions yet: the caller +// still needs the reader's local-priority and delivery-time policy integration. +func run_remote_speech_playback(ctx context.Context, speaker *configuredSpeaker, + incoming <-chan tentacle.RemoteSpeechDelivery, allow func(tentacle.RemoteSpeechDelivery) bool) error { + if speaker == nil || allow == nil { + return errors.New("Remote speech playback requires a backend and delivery policy") + } + var voice *remoteSpeechVoice // Only the renderer accesses it until run joins. + var voiceLease tentaclewire.RemoteLease + var stopLease func() bool + closeVoice := func() { + if stopLease != nil { + stopLease() + } + if voice != nil { + cleanup, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = voice.Close(cleanup) + } + } + defer closeVoice() + return run_remote_speech_stream(ctx, incoming, allow, func(request context.Context, delivery tentacle.RemoteSpeechDelivery, utterance remoteSpeechUtterance) error { + if voice == nil || voice.ctx.Err() != nil || voiceLease != delivery.Lease { + closeVoice() + var err error + voice, err = speaker.open_remote_voice(request) + if err != nil { + return err + } + voiceLease = delivery.Lease + owned := voice + // Even an idle handle belongs to its original lease. Close it on + // revoke, on a backend worker rather than the policy/input caller. + stopLease = context.AfterFunc(delivery.Context, func() { + cleanup, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = owned.Close(cleanup) + }) + } + // Connecting/querying the backend can take time. Admission cannot + // substitute for the original authority at synthesis handoff. + if request.Err() != nil { + return request.Err() + } + if !allow(delivery) { + return errors.New("Remote speech delivery policy changed") + } + return voice.SpeakUtteranceAndWait(request, utterance) + }) +} + +// run_remote_speech_stream is an unwired reader-owner prerequisite. allow must +// recheck original lease, focus and output policy without backend I/O. render +// must honor cancellation, recheck policy after acquisition, and complete scoped +// backend cleanup before returning. Neither callback may observe this output. +// The caller must retire the session on error. A dequeue retains mailbox budget +// until completion/discard, including while a cancelled renderer is closing. +func run_remote_speech_stream(ctx context.Context, incoming <-chan tentacle.RemoteSpeechDelivery, + allow func(tentacle.RemoteSpeechDelivery) bool, + render func(context.Context, tentacle.RemoteSpeechDelivery, remoteSpeechUtterance) error) error { + if allow == nil || render == nil { + return errors.New("Remote speech owner requires delivery policy and renderer") + } + var pending []*remoteSpeechStreamItem + var active *remoteSpeechStreamItem + var finished chan error + var cancel context.CancelFunc + 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) + } + release := func(item *remoteSpeechStreamItem) { + items -= len(item.delivery.Speeches) + bytes -= item.bytes + item.delivery.Release() + } + discard := func() { + for _, item := range pending { + release(item) + } + pending = nil + } + interrupt := func() { + discard() + if cancel != nil { + cancel() + } + } + defer func() { + interrupt() + if active != nil { + <-finished // Renderer owns cleanup; no replacement can overtake it. + release(active) + } + }() + for { + if ctx.Err() != nil { + return ctx.Err() + } + for active == nil && len(pending) != 0 { + item := pending[0] + pending[0] = nil + pending = pending[1:] + if !valid(item.delivery) { + release(item) + continue + } + request, stop := context.WithCancel(item.delivery.Context) + cancel, active, activeContext, finished = stop, item, request, make(chan error, 1) + go func(result chan<- error) { result <- render(request, item.delivery, item.utterance) }(finished) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-leaseDone: + interrupt() + leaseDone = nil + case err := <-finished: + // A deliberate stream interruption is not a backend failure. Never + // retry its suffix; the renderer may have spoken a prefix already. + interrupted := activeContext.Err() != nil + cancel() + release(active) + active, cancel, finished = nil, nil, nil + if err != nil && !interrupted { + return errors.New("Remote speech renderer failed") + } + case delivery, ok := <-incoming: + if !ok { + return nil + } + if delivery.Release == nil { + return errors.New("Remote speech delivery has no budget owner") + } + if !valid(delivery) { + delivery.Release() + continue + } + if lease != delivery.Lease { + interrupt() + lease, leaseDone, paused = delivery.Lease, delivery.Context.Done(), false + } + if len(delivery.Speeches) == 1 { + effect := delivery.Speeches[0] + if tentaclewire.ValidRemoteSpeech(effect) && (effect.Kind == "cancel" || effect.Kind == "pause") { + if effect.Kind == "cancel" || *effect.Paused { + interrupt() + } + if effect.Kind == "pause" { + paused = *effect.Paused + } + delivery.Release() + continue + } + } + utterance, err := prepare_remote_utterance(delivery.Speeches) + if err != nil { + delivery.Release() + return err + } + if paused { + delivery.Release() + continue + } + item := &remoteSpeechStreamItem{delivery: delivery, utterance: utterance} + for _, frame := range delivery.Speeches { + encoded, _ := json.Marshal(frame) + item.bytes += len(encoded) + } + if items+len(delivery.Speeches) > tentaclewire.RemoteSpeechQueueSize || bytes+item.bytes > tentaclewire.RemoteSpeechQueueBytes { + delivery.Release() + return errors.New("Remote speech owner budget exceeded") + } + items += len(delivery.Speeches) + bytes += item.bytes + // NEXT waits for the current assembly. NOW interrupts only lower + // priority work; equal priorities remain FIFO. Interrupted speech + // is discarded, because this backend cannot resume without replay. + if utterance.priority == 2 && active != nil && active.utterance.priority < 2 { + cancel() + } + position := len(pending) + for position > 0 && pending[position-1].utterance.priority < utterance.priority { + position-- + } + pending = append(pending, nil) + copy(pending[position+1:], pending[position:]) + pending[position] = item + } + } +} diff --git a/internal/app/remote_speech_stream_test.go b/internal/app/remote_speech_stream_test.go new file mode 100644 index 0000000..13a4e30 --- /dev/null +++ b/internal/app/remote_speech_stream_test.go @@ -0,0 +1,447 @@ +// Copyright (c) 2026 Storm Dragon +// Authors: Storm Dragon, Angus Kola +// SPDX-License-Identifier: GPL-3.0-or-later + +package app + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "git.stormux.org/storm/cthulhu/internal/tentacle" + "git.stormux.org/storm/cthulhu/internal/tentaclewire" +) + +type remoteStreamCall struct { + ctx context.Context + utterance remoteSpeechUtterance + finish chan error + stopped chan struct{} + cleanup chan struct{} +} + +type remoteStreamFixture struct { + incoming chan tentacle.RemoteSpeechDelivery + started chan remoteStreamCall + done chan error + ctx context.Context + cancel context.CancelFunc + lease tentaclewire.RemoteLease + hold bool + allowed atomic.Bool +} + +func remote_stream_fixture(t *testing.T, hold bool) *remoteStreamFixture { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + f := &remoteStreamFixture{ctx: ctx, cancel: cancel, hold: hold, + incoming: make(chan tentacle.RemoteSpeechDelivery, 128), started: make(chan remoteStreamCall, 128), done: make(chan error, 1), + lease: tentaclewire.RemoteLease{Lease: strings.Repeat("a", 32), Epoch: 1}} + f.allowed.Store(true) + go func() { + f.done <- run_remote_speech_stream(ctx, f.incoming, func(tentacle.RemoteSpeechDelivery) bool { return f.allowed.Load() }, + func(ctx context.Context, _ tentacle.RemoteSpeechDelivery, utterance remoteSpeechUtterance) error { + call := remoteStreamCall{ctx: ctx, utterance: utterance, finish: make(chan error, 1), stopped: make(chan struct{}), cleanup: make(chan struct{})} + f.started <- call + select { + case err := <-call.finish: + return err + case <-ctx.Done(): + close(call.stopped) + if hold { + <-call.cleanup + } + return ctx.Err() + } + }) + }() + t.Cleanup(cancel) + return f +} + +func remote_stream_wait(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Remote stream did not reach its synchronization point") + } +} + +func (f *remoteStreamFixture) send(t *testing.T, effect tentaclewire.RemoteSpeech) <-chan struct{} { + t.Helper() + return f.send_delivery(t, f.ctx, f.lease, []tentaclewire.RemoteSpeech{effect}) +} + +func (f *remoteStreamFixture) send_delivery(t *testing.T, ctx context.Context, lease tentaclewire.RemoteLease, frames []tentaclewire.RemoteSpeech) <-chan struct{} { + t.Helper() + released := make(chan struct{}) + var calls atomic.Int32 + f.incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease, Speeches: frames, Release: func() { + if calls.Add(1) != 1 { + t.Error("Remote stream released mailbox budget twice") + return + } + close(released) + }} + return released +} + +func (f *remoteStreamFixture) speak(t *testing.T, text string, priority int) <-chan struct{} { + return f.send(t, remote_render_frame(priority, tentaclewire.RemoteSpeechPart{Text: text})) +} + +func (f *remoteStreamFixture) next(t *testing.T, text string) remoteStreamCall { + t.Helper() + select { + case call := <-f.started: + if len(call.utterance.fragments) != 1 || call.utterance.fragments[0].Text != text { + t.Fatalf("unexpected Remote rendering: %#v", call.utterance) + } + return call + case <-time.After(3 * time.Second): + t.Fatal("Remote stream did not start rendering") + return remoteStreamCall{} + } +} + +func (f *remoteStreamFixture) barrier(t *testing.T) { + t.Helper() + paused := false + remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "pause", Paused: &paused})) +} + +func (f *remoteStreamFixture) stop(t *testing.T) { + t.Helper() + f.cancel() + err := remote_render_result(t, f.done) + if !errors.Is(err, context.Canceled) { + t.Fatalf("stream shutdown = %v", err) + } +} + +func TestRemoteSpeechStreamPriorityAndCleanup(t *testing.T) { + f := remote_stream_fixture(t, true) + firstRelease := f.speak(t, "active", 0) + first := f.next(t, "active") + normalRelease := f.speak(t, "normal", 0) + nextRelease := f.speak(t, "next", 1) + f.barrier(t) + select { + case <-first.stopped: + t.Fatal("NEXT interrupted the current assembly") + default: + } + nowRelease := f.speak(t, "now", 2) + remote_stream_wait(t, first.stopped) + f.barrier(t) + select { + case <-firstRelease: + t.Fatal("active budget was freed before renderer cleanup") + case <-f.started: + t.Fatal("replacement rendering overtook cleanup") + default: + } + close(first.cleanup) + now := f.next(t, "now") + remote_stream_wait(t, firstRelease) + secondNowRelease := f.speak(t, "second now", 2) + f.barrier(t) + select { + case <-now.stopped: + t.Fatal("NOW interrupted equal-priority speech") + default: + } + now.finish <- nil + second := f.next(t, "second now") + second.finish <- nil + next := f.next(t, "next") + next.finish <- nil + normal := f.next(t, "normal") + normal.finish <- nil + for _, release := range []<-chan struct{}{normalRelease, nextRelease, nowRelease, secondNowRelease} { + remote_stream_wait(t, release) + } + f.stop(t) +} + +func TestRemoteSpeechStreamCancelAndPause(t *testing.T) { + for _, kind := range []string{"cancel", "pause"} { + t.Run(kind, func(t *testing.T) { + f := remote_stream_fixture(t, true) + activeRelease := f.speak(t, "active", 0) + active := f.next(t, "active") + pendingRelease := f.speak(t, "discard pending", 0) + effect := tentaclewire.RemoteSpeech{Kind: kind} + if kind == "pause" { + paused := true + effect.Paused = &paused + } + 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") + remote_stream_wait(t, activeRelease) + future.finish <- nil + remote_stream_wait(t, futureRelease) + f.stop(t) + }) + } +} + +func TestRemoteSpeechStreamRevocationAndReplacement(t *testing.T) { + f := remote_stream_fixture(t, true) + leaseContext, revoke := context.WithCancel(f.ctx) + firstRelease := f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "old"})}) + first := f.next(t, "old") + pendingRelease := f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "old pending"})}) + f.barrier(t) + revoke() + remote_stream_wait(t, first.stopped) + remote_stream_wait(t, pendingRelease) + remote_stream_wait(t, f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(2, tentaclewire.RemoteSpeechPart{Text: "stale"})})) + f.lease.Epoch++ + newRelease := f.speak(t, "replacement", 0) + f.barrier(t) + close(first.cleanup) + replacement := f.next(t, "replacement") + remote_stream_wait(t, firstRelease) + replacement.finish <- nil + remote_stream_wait(t, newRelease) + f.stop(t) +} + +func TestRemoteSpeechStreamEOFRetainsBudgetThroughCleanup(t *testing.T) { + f := remote_stream_fixture(t, true) + activeRelease := f.speak(t, "active", 0) + active := f.next(t, "active") + pendingRelease := f.speak(t, "pending", 0) + f.barrier(t) + close(f.incoming) + remote_stream_wait(t, active.stopped) + remote_stream_wait(t, pendingRelease) + select { + case <-activeRelease: + t.Fatal("EOF released active budget before cleanup") + case <-f.done: + t.Fatal("EOF returned before renderer cleanup") + default: + } + close(active.cleanup) + if err := remote_render_result(t, f.done); err != nil { + t.Fatal(err) + } + remote_stream_wait(t, activeRelease) +} + +func TestRemoteSpeechStreamRechecksPolicyAndFailsClosed(t *testing.T) { + for _, cause := range []string{"policy", "backend", "invalid", "overflow", "bytes"} { + t.Run(cause, func(t *testing.T) { + f := remote_stream_fixture(t, false) + activeRelease := f.speak(t, "active", 0) + active := f.next(t, "active") + pendingRelease := f.speak(t, "pending", 0) + f.barrier(t) + switch cause { + case "policy": + f.allowed.Store(false) + active.finish <- nil + remote_stream_wait(t, pendingRelease) + remote_stream_wait(t, f.speak(t, "denied", 2)) + f.stop(t) + case "backend": + active.finish <- errors.New("backend failure") + if err := remote_render_result(t, f.done); err == nil || errors.Is(err, context.Canceled) { + t.Fatalf("backend failure did not retire the stream: %v", err) + } + case "invalid": + remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "speak"})) + // Release precedes error return; wait for the owned result. + if err := remote_render_result(t, f.done); err == nil { + t.Fatal("invalid assembly was accepted") + } + case "overflow": + for range tentaclewire.RemoteSpeechQueueSize - 2 { + f.speak(t, "bounded", 0) + } + remote_stream_wait(t, f.speak(t, "overflow", 0)) + if err := remote_render_result(t, f.done); err == nil { + t.Fatal("owner exceeded retained frame budget") + } + case "bytes": + // Frame count remains below 64 while encoded bytes exceed 256 KiB. + for range 52 { + f.speak(t, strings.Repeat("x", 5000), 0) + } + remote_stream_wait(t, f.speak(t, strings.Repeat("x", 5000), 0)) + if err := remote_render_result(t, f.done); err == nil { + t.Fatal("owner exceeded retained byte budget") + } + } + remote_stream_wait(t, activeRelease) + remote_stream_wait(t, pendingRelease) + }) + } +} + +func TestRemoteSpeechStreamIsolatedBackendCancellation(t *testing.T) { + speaker, server := isolated_voice_fixture(t) + 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 }) + }() + lease := tentaclewire.RemoteLease{Lease: strings.Repeat("b", 32), Epoch: 1} + send := func(effect tentaclewire.RemoteSpeech) <-chan struct{} { + released := make(chan struct{}) + incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease, Speeches: []tentaclewire.RemoteSpeech{effect}, Release: func() { close(released) }} + return released + } + localDone := make(chan error, 1) + go func() { localDone <- speaker.SpeakAndWait(ctx, "local status", nil) }() + local := server.next_utterance(t) + remoteRelease := send(remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "remote first"})) + remote := server.next_utterance(t) + if local.id == remote.id { + t.Fatal("stream acquired the local speaking client") + } + remote_stream_wait(t, send(tentaclewire.RemoteSpeech{Kind: "cancel"})) + remote_stream_wait(t, remoteRelease) + server.mutex.Lock() + localPending := server.connections[local.id].pending + server.mutex.Unlock() + if localPending == "" { + t.Fatal("remote cancel removed local status") + } + nextRelease := send(remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "remote after cancel"})) + next := server.next_utterance(t) + if next.text != "remote after cancel" || next.id == local.id || next.id == remote.id { + t.Fatal("cancelled renderer was reused or replayed") + } + server.end_utterance(t, next) + remote_stream_wait(t, nextRelease) + server.end_utterance(t, local) + if err := remote_render_result(t, localDone); err != nil { + t.Fatal(err) + } + 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{}) + server.mutex.Lock() + server.holdCommand, server.held, server.resume = "HISTORY GET CLIENT_ID", held, resume + server.mutex.Unlock() + // Unblock the private server even if an assertion fails. + defer func() { + select { + case <-resume: + default: + close(resume) + } + }() + var allowed atomic.Bool + allowed.Store(true) + incoming := make(chan tentacle.RemoteSpeechDelivery, 1) + released := make(chan struct{}) + incoming <- tentacle.RemoteSpeechDelivery{Context: t.Context(), + Lease: tentaclewire.RemoteLease{Lease: strings.Repeat("c", 32), Epoch: 1}, + Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "denied after acquisition"})}, + Release: func() { close(released) }} + done := make(chan error, 1) + go func() { + done <- run_remote_speech_playback(t.Context(), speaker, incoming, func(tentacle.RemoteSpeechDelivery) bool { return allowed.Load() }) + }() + remote_stream_wait(t, held) + allowed.Store(false) + server.mutex.Lock() + server.holdCommand = "" + server.mutex.Unlock() + close(resume) + if err := remote_render_result(t, done); err == nil { + t.Fatal("policy loss during acquisition was accepted") + } + remote_stream_wait(t, released) + select { + case <-server.queued: + t.Fatal("policy loss exposed speech to the backend") + default: + } +} + +func TestRemoteSpeechStreamVoiceRetiresWithOriginalLease(t *testing.T) { + for _, state := range []string{"active", "idle"} { + t.Run(state, func(t *testing.T) { test_remote_stream_lease_voice(t, state == "active") }) + } +} + +func test_remote_stream_lease_voice(t *testing.T, active bool) { + t.Helper() + speaker, server := isolated_voice_fixture(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + leaseContext, revoke := context.WithCancel(ctx) + defer revoke() + incoming := make(chan tentacle.RemoteSpeechDelivery, 4) + done := make(chan error, 1) + go func() { + done <- run_remote_speech_playback(ctx, speaker, incoming, func(delivery tentacle.RemoteSpeechDelivery) bool { return delivery.Context.Err() == nil }) + }() + send := func(lifetime context.Context, epoch uint64, text string) <-chan struct{} { + released := make(chan struct{}) + incoming <- tentacle.RemoteSpeechDelivery{Context: lifetime, + Lease: tentaclewire.RemoteLease{Lease: strings.Repeat("d", 32), Epoch: epoch}, + Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: text})}, + Release: func() { close(released) }} + return released + } + firstRelease := send(leaseContext, 1, "first lease") + first := server.next_utterance(t) + if !active { + server.end_utterance(t, first) + remote_stream_wait(t, firstRelease) + } + speaker.mu.RLock() + oldVoice := speaker.remote + speaker.mu.RUnlock() + revoke() + remote_stream_wait(t, oldVoice.closed) + remote_stream_wait(t, firstRelease) + nextRelease := send(ctx, 2, "replacement lease") + next := server.next_utterance(t) + if next.id == first.id || next.text != "replacement lease" { + t.Fatal("replacement inherited the retired backend client") + } + // Repeated old cleanup cannot cancel the replacement's pending message. + if err := oldVoice.Close(t.Context()); err != nil { + t.Fatal(err) + } + server.mutex.Lock() + pending := server.connections[next.id].pending + server.mutex.Unlock() + if pending != "replacement lease" { + t.Fatal("old lease cleanup removed replacement speech") + } + server.end_utterance(t, next) + remote_stream_wait(t, nextRelease) + cancel() + if err := remote_render_result(t, done); !errors.Is(err, context.Canceled) { + t.Fatalf("stream shutdown: %v", err) + } +}