Add bounded incoming Remote speech stream ownership

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.

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
+		}
+	}
+}

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)
+	}
+}