ak47
49be8309fe
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.
438 lines
14 KiB
Go
438 lines
14 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// Authors: Storm Dragon, Angus Kola
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package app
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/config"
|
|
)
|
|
|
|
type isolatedVoiceConnection struct {
|
|
net.Conn
|
|
write sync.Mutex
|
|
// The following fields belong to isolatedVoiceServer.mutex.
|
|
module, pending string
|
|
priority string
|
|
notifications bool
|
|
language, spelling, ssml string
|
|
voiceType, synthesis string
|
|
}
|
|
|
|
type isolatedVoiceUtterance struct {
|
|
id int
|
|
text, language, spelling, ssml string
|
|
module, voiceType, synthesis string
|
|
priority string
|
|
}
|
|
|
|
func (connection *isolatedVoiceConnection) send(message string) {
|
|
connection.write.Lock()
|
|
defer connection.write.Unlock()
|
|
_, _ = fmt.Fprint(connection.Conn, message)
|
|
}
|
|
|
|
// Keep local and remote utterances outstanding until an explicit END/CANCEL.
|
|
// This models cancellation ownership, not synthesized audio or server priority.
|
|
type isolatedVoiceServer struct {
|
|
mutex sync.Mutex
|
|
connections map[int]*isolatedVoiceConnection
|
|
accepted int
|
|
queued chan isolatedVoiceUtterance
|
|
queryReplies map[string]string
|
|
holdCommand string
|
|
held, resume chan struct{}
|
|
}
|
|
|
|
func isolated_voice_fixture(t *testing.T) (*configuredSpeaker, *isolatedVoiceServer) {
|
|
t.Helper()
|
|
path := filepath.Join(t.TempDir(), "speech.sock")
|
|
listener, err := net.Listen("unix", path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
server := &isolatedVoiceServer{connections: make(map[int]*isolatedVoiceConnection), queued: make(chan isolatedVoiceUtterance, 128)}
|
|
var handlers sync.WaitGroup
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
for {
|
|
conn, err := listener.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
connection := &isolatedVoiceConnection{Conn: conn, module: "default", language: "en-GB", spelling: "off", ssml: "on", voiceType: "FEMALE2"}
|
|
server.mutex.Lock()
|
|
server.accepted++
|
|
id := server.accepted
|
|
server.connections[id] = connection
|
|
server.mutex.Unlock()
|
|
handlers.Add(1)
|
|
go func() {
|
|
defer handlers.Done()
|
|
defer conn.Close()
|
|
reader := bufio.NewReader(conn)
|
|
for {
|
|
line, err := reader.ReadString('\n')
|
|
if err != nil {
|
|
return
|
|
}
|
|
command := strings.TrimSpace(line)
|
|
response := "200 OK\r\n"
|
|
var queued *isolatedVoiceUtterance
|
|
server.mutex.Lock()
|
|
switch {
|
|
case command == "HISTORY GET CLIENT_ID":
|
|
response = fmt.Sprintf("245-%d\r\n245 OK\r\n", id)
|
|
case command == "GET LANGUAGE":
|
|
response = "251-" + connection.language + "\r\n251 OK\r\n"
|
|
case command == "GET OUTPUT_MODULE":
|
|
response = "251-" + connection.module + "\r\n251 OK\r\n"
|
|
case command == "GET VOICE_TYPE":
|
|
response = "251-" + connection.voiceType + "\r\n251 OK\r\n"
|
|
case strings.HasPrefix(command, "SET SELF PRIORITY "):
|
|
connection.priority = strings.TrimPrefix(command, "SET SELF PRIORITY ")
|
|
case strings.HasPrefix(command, "SET SELF LANGUAGE "):
|
|
connection.language = strings.TrimPrefix(command, "SET SELF LANGUAGE ")
|
|
// Server LanguageDefaultModule mappings must not silently
|
|
// replace the reader's chosen synthesis module.
|
|
connection.module = "language-mapped"
|
|
if connection.language == "rejected" {
|
|
response = "400 Rejected\r\n"
|
|
}
|
|
case strings.HasPrefix(command, "SET SELF SPELLING "):
|
|
connection.spelling = strings.TrimPrefix(command, "SET SELF SPELLING ")
|
|
case strings.HasPrefix(command, "SET SELF SSML_MODE "):
|
|
connection.ssml = strings.TrimPrefix(command, "SET SELF SSML_MODE ")
|
|
case strings.HasPrefix(command, "SET SELF VOICE_TYPE "):
|
|
connection.voiceType = strings.TrimPrefix(command, "SET SELF VOICE_TYPE ")
|
|
connection.synthesis = ""
|
|
case strings.HasPrefix(command, "SET SELF SYNTHESIS_VOICE "):
|
|
connection.synthesis = strings.TrimPrefix(command, "SET SELF SYNTHESIS_VOICE ")
|
|
connection.voiceType = "NO_VOICE"
|
|
case strings.HasPrefix(command, "SET SELF OUTPUT_MODULE "):
|
|
connection.module = strings.TrimPrefix(command, "SET SELF OUTPUT_MODULE ")
|
|
connection.synthesis = ""
|
|
if connection.module == "rejected" {
|
|
response = "400 Rejected\r\n"
|
|
}
|
|
case strings.HasPrefix(command, "SET SELF NOTIFICATION ALL "):
|
|
connection.notifications = strings.HasSuffix(command, " on")
|
|
case strings.HasPrefix(command, "CANCEL "):
|
|
target, _ := strconv.Atoi(strings.TrimPrefix(command, "CANCEL "))
|
|
speaking := server.connections[target]
|
|
if speaking == nil {
|
|
server.mutex.Unlock()
|
|
t.Error("cancellation did not target a speaking client")
|
|
return
|
|
}
|
|
notify := speaking.notifications && speaking.pending != ""
|
|
speaking.pending = ""
|
|
server.mutex.Unlock()
|
|
connection.send(response)
|
|
if notify {
|
|
speaking.send(fmt.Sprintf("703-1\r\n703-%d\r\n703 CANCELED\r\n", target))
|
|
}
|
|
continue
|
|
case command == "SPEAK":
|
|
server.mutex.Unlock()
|
|
var lines []string
|
|
for {
|
|
part, err := reader.ReadString('\n')
|
|
if err != nil {
|
|
return
|
|
}
|
|
if part == ".\r\n" {
|
|
break
|
|
}
|
|
part = strings.TrimSuffix(part, "\r\n")
|
|
if strings.HasPrefix(part, "..") {
|
|
part = part[1:]
|
|
}
|
|
lines = append(lines, part)
|
|
}
|
|
server.mutex.Lock()
|
|
connection.pending = strings.Join(lines, "\n")
|
|
queued = &isolatedVoiceUtterance{id: id, text: connection.pending, language: connection.language, spelling: connection.spelling, ssml: connection.ssml, module: connection.module, voiceType: connection.voiceType, synthesis: connection.synthesis, priority: connection.priority}
|
|
response = "230 OK\r\n225 QUEUED\r\n"
|
|
if connection.notifications {
|
|
response += fmt.Sprintf("701-1\r\n701-%d\r\n701 BEGIN\r\n", id)
|
|
}
|
|
}
|
|
if override := server.queryReplies[command]; override != "" {
|
|
response = override
|
|
}
|
|
hold := command == server.holdCommand
|
|
held, resume := server.held, server.resume
|
|
server.mutex.Unlock()
|
|
if hold {
|
|
close(held)
|
|
<-resume
|
|
}
|
|
connection.send(response)
|
|
if queued != nil {
|
|
server.queued <- *queued
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
}()
|
|
settings := config.DefaultSettings()
|
|
settings.UseCustomEchoVoice = true
|
|
settings.UseCustomEchoSpeechServer = true
|
|
settings.EchoSpeechServer = "echo"
|
|
speaker := newConfiguredSpeaker(path, settings)
|
|
t.Cleanup(func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
if err := speaker.Close(ctx); err != nil {
|
|
t.Error(err)
|
|
}
|
|
_ = listener.Close()
|
|
<-done
|
|
server.mutex.Lock()
|
|
for _, conn := range server.connections {
|
|
_ = conn.Close()
|
|
}
|
|
server.mutex.Unlock()
|
|
handlers.Wait()
|
|
})
|
|
if err := speaker.Connect(t.Context()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return speaker, server
|
|
}
|
|
|
|
func (server *isolatedVoiceServer) pending(text string) (int, string) {
|
|
server.mutex.Lock()
|
|
defer server.mutex.Unlock()
|
|
for id, conn := range server.connections {
|
|
if conn.pending == text {
|
|
return id, conn.module
|
|
}
|
|
}
|
|
return 0, ""
|
|
}
|
|
|
|
func begin_isolated_speech(t *testing.T, voice *remoteSpeechVoice, ctx context.Context, text string) <-chan error {
|
|
t.Helper()
|
|
began := make(chan struct{})
|
|
finished := make(chan error, 1)
|
|
go func() { finished <- voice.SpeakAndWait(ctx, text, func() { close(began) }) }()
|
|
select {
|
|
case <-began:
|
|
case err := <-finished:
|
|
t.Fatal("speech failed before BEGIN", err)
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("speech did not begin")
|
|
}
|
|
return finished
|
|
}
|
|
|
|
func finish_isolated_speech(t *testing.T, finished <-chan error) {
|
|
t.Helper()
|
|
select {
|
|
case err := <-finished:
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatal("speech was not cancelled", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("speech did not retire")
|
|
}
|
|
}
|
|
|
|
func TestRemoteSpeechVoiceCancellationKeepsLocalStatusAndEcho(t *testing.T) {
|
|
for _, operation := range []string{"cancel", "caller", "close"} {
|
|
t.Run(operation, func(t *testing.T) {
|
|
speaker, server := isolated_voice_fixture(t)
|
|
if err := speaker.SpeakVoice(t.Context(), "local status", "system"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := speaker.SpeakEcho(t.Context(), "local echo"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
voice, err := speaker.open_remote_voice(t.Context())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
defer cancel()
|
|
finished := begin_isolated_speech(t, voice, ctx, "remote")
|
|
remoteID, _ := server.pending("remote")
|
|
statusID, _ := server.pending("local status")
|
|
echoID, _ := server.pending("local echo")
|
|
if remoteID == 0 || statusID == 0 || echoID == 0 || remoteID == statusID || remoteID == echoID {
|
|
t.Fatal("Remote shared a speaking connection with local output")
|
|
}
|
|
select {
|
|
case <-finished:
|
|
t.Fatal("queue acknowledgement completed remote speech")
|
|
default:
|
|
}
|
|
switch operation {
|
|
case "cancel":
|
|
err = voice.Cancel(t.Context())
|
|
case "caller":
|
|
cancel()
|
|
case "close":
|
|
err = voice.Close(t.Context())
|
|
}
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish_isolated_speech(t, finished)
|
|
if id, _ := server.pending("local status"); id != statusID {
|
|
t.Fatal("remote cancellation discarded local status")
|
|
}
|
|
if id, _ := server.pending("local echo"); id != echoID {
|
|
t.Fatal("remote cancellation discarded custom echo")
|
|
}
|
|
if id, _ := server.pending("remote"); id != 0 {
|
|
t.Fatal("remote speech survived cancellation")
|
|
}
|
|
if operation != "close" {
|
|
// Notification cleanup leaves the same isolated connection reusable.
|
|
finished = begin_isolated_speech(t, voice, t.Context(), "remote again")
|
|
if err := speaker.Cancel(t.Context()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish_isolated_speech(t, finished)
|
|
if id, _ := server.pending("local status"); id != 0 {
|
|
t.Fatal("core cancellation failed to stop local output")
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRemoteSpeechVoiceConfigurationAndOwnership(t *testing.T) {
|
|
speaker, server := isolated_voice_fixture(t)
|
|
voice, err := speaker.open_remote_voice(t.Context())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := speaker.open_remote_voice(t.Context()); err == nil {
|
|
t.Fatal("two simultaneous Remote voices admitted")
|
|
}
|
|
finished := begin_isolated_speech(t, voice, t.Context(), "old settings")
|
|
settings := config.DefaultSettings()
|
|
settings.SpeechOutputModule = "rejected"
|
|
if err := speaker.UpdateConfiguration(t.Context(), settings); err == nil {
|
|
t.Fatal("rejected configuration committed")
|
|
}
|
|
if id, _ := server.pending("old settings"); id == 0 || voice.ctx.Err() != nil {
|
|
t.Fatal("failed preparation cancelled the existing Remote voice")
|
|
}
|
|
settings.SpeechOutputModule = "replacement"
|
|
if err := speaker.UpdateConfiguration(t.Context(), settings); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish_isolated_speech(t, finished)
|
|
if err := voice.SpeakAndWait(t.Context(), "stale", nil); !errors.Is(err, context.Canceled) {
|
|
t.Fatal("retired handle reused new configuration", err)
|
|
}
|
|
replacement, err := speaker.open_remote_voice(t.Context())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finished = begin_isolated_speech(t, replacement, t.Context(), "new settings")
|
|
if _, module := server.pending("new settings"); module != "replacement" {
|
|
t.Fatal("isolated voice used stale settings", module)
|
|
}
|
|
if err := voice.Close(t.Context()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if id, _ := server.pending("new settings"); id == 0 {
|
|
t.Fatal("stale close cancelled replacement voice")
|
|
}
|
|
if err := speaker.Close(t.Context()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
finish_isolated_speech(t, finished)
|
|
if _, err := speaker.open_remote_voice(t.Context()); err == nil {
|
|
t.Fatal("closed speaker admitted Remote")
|
|
}
|
|
}
|
|
|
|
func TestRemoteSpeechVoiceCompletionAndAcquisitionLifetime(t *testing.T) {
|
|
speaker, server := isolated_voice_fixture(t)
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
defer cancel()
|
|
server.mutex.Lock()
|
|
before := server.accepted
|
|
server.mutex.Unlock()
|
|
voice, err := speaker.open_remote_voice(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Acquisition cancellation must not become accidental session revocation.
|
|
cancel()
|
|
finished := begin_isolated_speech(t, voice, t.Context(), "complete")
|
|
server.mutex.Lock()
|
|
if server.accepted != before+2 {
|
|
t.Error("isolated voice did not open exactly one speaking/control pair")
|
|
}
|
|
var speaking *isolatedVoiceConnection
|
|
for _, connection := range server.connections {
|
|
if connection.pending == "complete" {
|
|
speaking = connection
|
|
connection.pending = ""
|
|
break
|
|
}
|
|
}
|
|
server.mutex.Unlock()
|
|
if speaking == nil {
|
|
t.Fatal("missing outstanding utterance")
|
|
}
|
|
speaking.send("702-1\r\n702-1\r\n702 END\r\n")
|
|
select {
|
|
case err := <-finished:
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("END did not complete speech")
|
|
}
|
|
if err := voice.Close(t.Context()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := speaker.open_remote_voice(t.Context()); err != nil {
|
|
t.Fatal("closed voice retained exclusive ownership", err)
|
|
}
|
|
}
|
|
|
|
func TestRemoteSpeechVoiceRejectsUnavailableBackendsAndCancelledAcquisition(t *testing.T) {
|
|
for _, active := range []configuredVoice{nil, &hardwareSpeaker{}} {
|
|
speaker := &configuredSpeaker{active: active}
|
|
if _, err := speaker.open_remote_voice(t.Context()); err == nil {
|
|
t.Fatal("unavailable software backend admitted Remote")
|
|
}
|
|
}
|
|
speaker, server := isolated_voice_fixture(t)
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
cancel()
|
|
server.mutex.Lock()
|
|
before := server.accepted
|
|
server.mutex.Unlock()
|
|
if _, err := speaker.open_remote_voice(ctx); !errors.Is(err, context.Canceled) {
|
|
t.Fatal("cancelled acquisition succeeded", err)
|
|
}
|
|
server.mutex.Lock()
|
|
defer server.mutex.Unlock()
|
|
if server.accepted != before {
|
|
t.Fatal("cancelled acquisition opened connections")
|
|
}
|
|
}
|