Enable isolated Remote speech and protect reading exports

Attach incoming software speech to reader sessions with lease checks, local output priority and isolated cancellation. Preserve every document fragment source for privacy checks and export reviewed browse current-line reading, spelling and phonetic output.

Verified focused reading and Remote race tests, complete review and external Remote race suites, private relay integration, speech regressions, affected-package vet, reader build and Tentacles make check. Audible speech, physical input/Braille and live interoperability remain unverified.

internal/app/tentacle_remote_speech_test.go

@@ -0,0 +1,131 @@
+// Copyright (c) 2026 Storm Dragon
+// Authors: Storm Dragon, Angus Kola
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app
+
+import (
+	"context"
+	"log/slog"
+	"testing"
+	"time"
+
+	"git.stormux.org/storm/cthulhu/internal/tentacle"
+)
+
+func TestRemoteSpeechLocalPriorityDiscardsActiveAndQueuedWork(t *testing.T) {
+	for _, change := range []string{"announcement", "interrupt", "mode", "disable", "secure"} {
+		t.Run(change, func(t *testing.T) {
+			worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
+			worker.running, worker.workerContext = true, t.Context()
+			f := remote_stream_fixture(t, false, worker.admit_remote_speech)
+			activeRelease := f.speak(t, "old active", 2)
+			active := f.next(t, "old active")
+			pendingRelease := f.speak(t, "old pending", 2)
+			// Equal priority keeps the queued item behind this active NOW.
+			// Use the barrier to establish admission before local output.
+			f.barrier(t)
+			switch change {
+			case "announcement":
+				worker.Enqueue(speechRequest{Text: "local status"})
+			case "interrupt":
+				worker.InterruptForUser()
+			case "mode":
+				worker.cycle_speech_mode()
+			case "disable":
+				worker.SetEnabled(false)
+			case "secure":
+				worker.SetSecureInputActive(true)
+			}
+			remote_stream_wait(t, active.stopped)
+			remote_stream_wait(t, activeRelease)
+			remote_stream_wait(t, pendingRelease)
+			remote_stream_wait(t, f.speak(t, "blocked by local policy", 2))
+			select {
+			case <-f.started:
+				t.Fatal("old or blocked incoming speech reached rendering")
+			default:
+			}
+			worker.mutex.Lock()
+			worker.pending, worker.interrupting = nil, false
+			worker.enabled, worker.secure, worker.mode = true, false, speechModeSpeech
+			worker.mutex.Unlock()
+			freshRelease := f.speak(t, "fresh output", 0)
+			fresh := f.next(t, "fresh output")
+			fresh.finish <- nil
+			remote_stream_wait(t, freshRelease)
+			f.stop(t)
+		})
+	}
+}
+
+func TestRemoteSpeechLocalPriorityRejectsOldMailboxAndIncompleteAssembly(t *testing.T) {
+	worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
+	worker.running, worker.workerContext = true, t.Context()
+	// The first fragment's timestamp survives finalization: work retained in
+	// the IPC assembly or mailbox must not acquire fresh local authority later.
+	old := tentacle.RemoteSpeechDelivery{Context: t.Context(), Received: time.Now()}
+	worker.Enqueue(speechRequest{Text: "local status"})
+	worker.mutex.Lock()
+	worker.pending = nil // Local queue drained before the old mailbox is read.
+	worker.mutex.Unlock()
+	if _, ok := worker.admit_remote_speech(old); ok {
+		t.Fatal("old mailbox/assembly work followed local output")
+	}
+	fresh := tentacle.RemoteSpeechDelivery{Context: t.Context(), Received: time.Now()}
+	permit, ok := worker.admit_remote_speech(fresh)
+	if !ok || permit.context.Err() != nil || !permit.current() {
+		t.Fatal("fresh incoming output was rejected after local reading")
+	}
+	permit.release()
+}
+
+func TestRemoteSpeechRejectedLocalRequestPreservesRemoteAndServerCancelContinues(t *testing.T) {
+	worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
+	worker.running, worker.workerContext = true, t.Context()
+	f := remote_stream_fixture(t, false, worker.admit_remote_speech)
+	released := f.speak(t, "active", 0)
+	active := f.next(t, "active")
+	oldGeneration := worker.generation + 1
+	worker.Enqueue(speechRequest{Text: "stale result", presentationGeneration: &oldGeneration})
+	select {
+	case <-active.ctx.Done():
+		t.Fatal("rejected local request cancelled Remote output")
+	default:
+	}
+	// A server priority cancellation discards this assembly. It must not
+	// disable the owner or retire consent, and no suffix is retried.
+	active.finish <- context.Canceled
+	remote_stream_wait(t, released)
+	nextRelease := f.speak(t, "after cancellation", 0)
+	next := f.next(t, "after cancellation")
+	next.finish <- nil
+	remote_stream_wait(t, nextRelease)
+	f.stop(t)
+}
+
+func TestRemoteSpeechBackgroundPriorityRejectionClosesCandidate(t *testing.T) {
+	speaker, server := isolated_voice_fixture(t)
+	server.mutex.Lock()
+	server.queryReplies = map[string]string{"SET SELF PRIORITY TEXT": "400 Rejected\r\n"}
+	server.mutex.Unlock()
+	if _, err := speaker.open_remote_voice(t.Context()); err == nil {
+		t.Fatal("isolated client acquired without lower backend priority")
+	}
+	speaker.mu.RLock()
+	remote := speaker.remote
+	speaker.mu.RUnlock()
+	if remote != nil {
+		t.Fatal("failed candidate became the owned voice")
+	}
+	server.mutex.Lock()
+	server.queryReplies = nil
+	server.mutex.Unlock()
+	voice, err := speaker.open_remote_voice(t.Context())
+	if err != nil {
+		t.Fatal("failed priority setup prevented a fresh acquisition", err)
+	}
+	if err := voice.Close(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+}