Add isolated Remote speech synthesis and typed rendering

Preserve literal whitespace and resolved spelling/language state across complete Remote speech assemblies. Isolate cancellation from local status and echo, retire handles on configuration changes or rendering failure, and cover metadata preparation cancellation with private SSIP fixtures.

Verified focused application and backend race tests, ten repetitions of isolated-client/rendering cases, shared speech regressions, affected-package vet, reader build and Tentacles checks. Incoming mailbox integration, local priority and live audible acceptance remain pending.
This commit is contained in:
ak47 committed 2026-11-04 19:56:38 -05:00
1 parent 622d2f96a1
commit ac431dd514
9 files changed
+1370 -11

No files matched your search

+25 -8
View File
@@ -31,6 +31,7 @@ type configuredSpeaker struct {
mu sync.RWMutex
update sync.Mutex
active configuredVoice
remote *remoteSpeechVoice
socket string
settings config.Settings
closed bool
@@ -70,10 +71,15 @@ func (speaker *configuredSpeaker) UpdateConfiguration(ctx context.Context, setti
}
hardware.settings = settings
hardware.mu.Unlock()
speaker.retire_remote_voice(settings)
return nil
}
if software, ok := current.(*echoSpeaker); ok && settings.SpeechServerFactory != "hardwarefactory" {
return software.UpdateConfiguration(ctx, settings)
if err := software.UpdateConfiguration(ctx, settings); err != nil {
return err
}
speaker.retire_remote_voice(settings)
return nil
}
var candidate configuredVoice
if settings.SpeechServerFactory == "hardwarefactory" {
@@ -99,8 +105,8 @@ func (speaker *configuredSpeaker) UpdateConfiguration(ctx context.Context, setti
}
}
speaker.active = candidate
speaker.settings = settings
speaker.mu.Unlock()
speaker.retire_remote_voice(settings)
// The new backend is committed. A retired connection's cleanup error must
// not report this successful swap as a rejected preferences transaction.
if current != nil {
@@ -182,23 +188,34 @@ func (speaker *configuredSpeaker) speak_uninterruptible(ctx context.Context, tex
func (speaker *configuredSpeaker) Cancel(ctx context.Context) error {
speaker.mu.RLock()
voice := speaker.active
remote := speaker.remote
speaker.mu.RUnlock()
if voice == nil {
return nil
var result error
if voice != nil {
result = voice.Cancel(ctx)
}
return voice.Cancel(ctx)
if remote != nil {
result = errors.Join(result, remote.Cancel(ctx))
}
return result
}
func (speaker *configuredSpeaker) Close(ctx context.Context) error {
speaker.update.Lock()
defer speaker.update.Unlock()
speaker.mu.Lock()
voice := speaker.active
remote := speaker.remote
speaker.active, speaker.closed = nil, true
speaker.remote = nil
speaker.mu.Unlock()
if voice == nil {
return nil
var result error
if remote != nil {
result = remote.Close(ctx)
}
return voice.Close(ctx)
if voice == nil {
return result
}
return errors.Join(result, voice.Close(ctx))
}
type hardwareSpeaker struct {
+108
View File
@@ -0,0 +1,108 @@
// 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"
"strings"
"time"
"git.stormux.org/storm/cthulhu/internal/speech"
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
)
// One complete reader assembly. The future mailbox owner must apply priority
// once, recheck delivery policy and retain its budget until rendering finishes.
// This value grants no authority and owns no lease or mailbox budget.
type remoteSpeechUtterance struct {
priority int
fragments []speech.Fragment
}
func prepare_remote_utterance(frames []tentaclewire.RemoteSpeech) (remoteSpeechUtterance, error) {
var utterance remoteSpeechUtterance
if len(frames) == 0 || len(frames) > tentaclewire.RemoteSpeechQueueSize {
return utterance, errors.New("invalid Remote speech assembly")
}
utterance.priority = frames[0].Priority
bytes := 0
// Validate every frame and the combined budget before retaining text or
// exposing a prefix to synthesis. Cancel/pause belong to the stream owner.
for _, frame := range frames {
if !tentaclewire.ValidRemoteSpeech(frame) || frame.Kind != "speak" || frame.Priority != utterance.priority {
return remoteSpeechUtterance{}, errors.New("invalid Remote speech assembly")
}
encoded, _ := json.Marshal(frame)
bytes += len(encoded)
if bytes > tentaclewire.RemoteSpeechQueueBytes {
return remoteSpeechUtterance{}, errors.New("Remote speech assembly exceeds its budget")
}
}
var text strings.Builder
var state speech.Fragment
flush := func() {
if text.Len() != 0 {
state.Text = text.String()
utterance.fragments = append(utterance.fragments, state)
text.Reset()
}
}
for _, frame := range frames {
for _, part := range frame.Parts {
if part.End {
flush()
continue
}
if state.Characters != part.Characters || state.Language != part.Language {
flush()
}
state.Characters, state.Language = part.Characters, part.Language
text.WriteString(part.Text)
}
}
flush()
return utterance, nil
}
// SpeakUtteranceAndWait renders an already-assembled utterance. It does not
// authorize delivery or interpret NVDA stream priority. Its caller must do so
// before invoking it. The backend validates all literal fragments before I/O.
func (voice *remoteSpeechVoice) SpeakUtteranceAndWait(ctx context.Context, utterance remoteSpeechUtterance) error {
if !voice.render.TryLock() {
return errors.New("Remote utterance is already rendering")
}
defer voice.render.Unlock()
if err := voice.ctx.Err(); err != nil {
return err
}
request, cancel := context.WithCancel(voice.ctx)
stop := context.AfterFunc(ctx, cancel)
defer func() { stop(); cancel() }()
voice.request.Lock()
voice.renderCancel = cancel
voice.request.Unlock()
defer func() {
voice.request.Lock()
voice.renderCancel = nil
voice.request.Unlock()
}()
if err := ctx.Err(); err != nil {
return err
}
if len(utterance.fragments) == 0 {
return nil
}
err := voice.client.SpeakFragmentsAndWait(request, utterance.fragments)
if err != nil {
// A rejected setting, interrupted block or cancelled fragment cannot
// seed the next assembly with partially changed metadata or output.
cleanup, finish := context.WithTimeout(context.Background(), 2*time.Second)
defer finish()
_ = voice.Close(cleanup)
}
return err
}
+371
View File
@@ -0,0 +1,371 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
"git.stormux.org/storm/cthulhu/internal/config"
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
)
func remote_render_frame(priority int, parts ...tentaclewire.RemoteSpeechPart) tentaclewire.RemoteSpeech {
return tentaclewire.RemoteSpeech{Kind: "speak", Priority: priority, Parts: parts}
}
func (server *isolatedVoiceServer) next_utterance(t *testing.T) isolatedVoiceUtterance {
t.Helper()
select {
case utterance := <-server.queued:
return utterance
case <-time.After(3 * time.Second):
t.Fatal("synthesis did not receive a fragment")
return isolatedVoiceUtterance{}
}
}
func (server *isolatedVoiceServer) end_utterance(t *testing.T, utterance isolatedVoiceUtterance) {
t.Helper()
server.mutex.Lock()
connection := server.connections[utterance.id]
connection.pending = ""
server.mutex.Unlock()
connection.send(fmt.Sprintf("702-1\r\n702-%d\r\n702 END\r\n", utterance.id))
}
func remote_render_result(t *testing.T, finished <-chan error) error {
t.Helper()
select {
case err := <-finished:
return err
case <-time.After(3 * time.Second):
t.Fatal("rendering did not finish")
return nil
}
}
func TestRemoteSpeechRenderingPreservesAssemblyAndLiteralMetadata(t *testing.T) {
speaker, server := isolated_voice_fixture(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
frames := []tentaclewire.RemoteSpeech{
remote_render_frame(2, tentaclewire.RemoteSpeechPart{Text: " hello"}),
remote_render_frame(2,
tentaclewire.RemoteSpeechPart{Text: " world\t"},
tentaclewire.RemoteSpeechPart{Text: "aa", Characters: true},
tentaclewire.RemoteSpeechPart{Text: "é\n.\nQUIT\r\n..literal\rend <speak> ", Language: "fr"},
tentaclewire.RemoteSpeechPart{End: true},
tentaclewire.RemoteSpeechPart{Text: " again "}),
}
utterance, err := prepare_remote_utterance(frames)
if err != nil {
t.Fatal(err)
}
if utterance.priority != 2 || len(utterance.fragments) != 4 {
t.Fatal("assembly lost priority or metadata boundaries")
}
finished := make(chan error, 1)
go func() { finished <- voice.SpeakUtteranceAndWait(t.Context(), utterance) }()
for index, expected := range []struct{ text, language, spelling string }{
{" hello world\t", "en-GB", "off"},
{"aa", "en-GB", "on"},
{"é\n.\nQUIT\n..literal\nend <speak> ", "fr", "off"},
{" again ", "en-GB", "off"},
} {
got := server.next_utterance(t)
if got.text != expected.text || got.language != expected.language || got.spelling != expected.spelling || got.ssml != "off" || got.module != "default" || got.voiceType != "FEMALE2" || got.synthesis != "" {
t.Fatalf("fragment %d = %#v; want %#v in plain-text mode", index, got, expected)
}
select {
case <-finished:
t.Fatal("queue acknowledgement finished assembly")
default:
}
server.end_utterance(t, got)
}
if err := remote_render_result(t, finished); err != nil {
t.Fatal(err)
}
// Successful rendering restores the effective server default and leaves
// the handle reusable. Transport splits are not synthesis boundaries.
large := strings.Repeat("界 ", 1200)
frames = nil
for _, part := range []string{large[:3000], large[3000:]} {
frames = append(frames, remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: part}))
}
utterance, err = prepare_remote_utterance(frames)
if err != nil {
t.Fatal(err)
}
go func() { finished <- voice.SpeakUtteranceAndWait(t.Context(), utterance) }()
got := server.next_utterance(t)
if got.text != large || got.language != "en-GB" || got.spelling != "off" {
t.Fatal("next assembly inherited state or lost split text")
}
server.end_utterance(t, got)
if err := remote_render_result(t, finished); err != nil {
t.Fatal(err)
}
}
func TestRemoteSpeechRenderingRestoresNamedVoice(t *testing.T) {
speaker, server := isolated_voice_fixture(t)
settings := speaker.settings
settings.DefaultVoice.Family = &config.VoiceFamily{Name: "configured voice"}
if err := speaker.UpdateConfiguration(t.Context(), settings); err != nil {
t.Fatal(err)
}
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0,
tentaclewire.RemoteSpeechPart{Text: "default"},
tentaclewire.RemoteSpeechPart{Text: "foreign", Language: "fr"},
tentaclewire.RemoteSpeechPart{Text: "restored"})})
if err != nil {
t.Fatal(err)
}
finished := make(chan error, 1)
go func() { finished <- voice.SpeakUtteranceAndWait(t.Context(), utterance) }()
for _, named := range []string{"configured voice", "", "configured voice"} {
got := server.next_utterance(t)
if got.synthesis != named || got.module != "default" || named == "" && got.voiceType != "MALE1" {
t.Fatalf("language switch did not scope the named voice: %#v", got)
}
server.end_utterance(t, got)
}
if err := remote_render_result(t, finished); err != nil {
t.Fatal(err)
}
}
func TestRemoteSpeechRenderingRejectsUnknownBackendDefaults(t *testing.T) {
for _, test := range []struct{ command, reply string }{
{"GET LANGUAGE", "400 Unsupported\r\n"},
{"GET LANGUAGE", "251-en GB\r\n251 OK\r\n"},
{"GET LANGUAGE", "251-en\r\n251-fr\r\n251 OK\r\n"},
{"GET OUTPUT_MODULE", "251-unsafe module\r\n251 OK\r\n"},
{"GET VOICE_TYPE", "251-unknown\r\n251 OK\r\n"},
{"GET VOICE_TYPE", "251-NO_VOICE\r\n251 OK\r\n"},
} {
t.Run(test.command+test.reply, func(t *testing.T) {
speaker, server := isolated_voice_fixture(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
server.mutex.Lock()
server.queryReplies = map[string]string{test.command: test.reply}
server.mutex.Unlock()
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "never queued"})})
if err != nil {
t.Fatal(err)
}
if err := voice.SpeakUtteranceAndWait(t.Context(), utterance); err == nil {
t.Fatal("unresolved backend defaults admitted")
}
if voice.ctx.Err() == nil {
t.Fatal("unresolved backend handle survived")
}
select {
case <-server.queued:
t.Fatal("unresolved defaults synthesized output")
default:
}
})
}
}
func TestRemoteSpeechRenderingCancellationDuringMetadata(t *testing.T) {
speaker, server := isolated_voice_fixture(t)
if err := speaker.SpeakVoice(t.Context(), "local status", "system"); err != nil {
t.Fatal(err)
}
server.next_utterance(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
held, resume := make(chan struct{}), make(chan struct{})
defer close(resume)
server.mutex.Lock()
server.holdCommand, server.held, server.resume = "GET LANGUAGE", held, resume
server.mutex.Unlock()
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "never queued"})})
if err != nil {
t.Fatal(err)
}
finished := make(chan error, 1)
go func() { finished <- voice.SpeakUtteranceAndWait(t.Context(), utterance) }()
select {
case <-held:
case <-time.After(3 * time.Second):
t.Fatal("metadata query did not start")
}
if err := voice.SpeakUtteranceAndWait(t.Context(), utterance); err == nil {
t.Fatal("overlapping typed assembly admitted")
}
if err := voice.Cancel(t.Context()); err != nil {
t.Fatal(err)
}
if err := remote_render_result(t, finished); !errors.Is(err, context.Canceled) {
t.Fatal("metadata cancellation was lost", err)
}
if id, _ := server.pending("local status"); id == 0 {
t.Fatal("metadata cancellation discarded local status")
}
select {
case <-server.queued:
t.Fatal("metadata cancellation synthesized output")
default:
}
}
func TestRemoteSpeechRenderingCancellationKeepsLocalOutputAndDropsSuffix(t *testing.T) {
for _, operation := range []string{"remote", "caller", "configuration"} {
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)
}
server.next_utterance(t)
if err := speaker.SpeakEcho(t.Context(), "local echo"); err != nil {
t.Fatal(err)
}
server.next_utterance(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0,
tentaclewire.RemoteSpeechPart{Text: "prefix", Characters: true, Language: "fr"},
tentaclewire.RemoteSpeechPart{End: true}, tentaclewire.RemoteSpeechPart{Text: "suffix"})})
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
finished := make(chan error, 1)
go func() { finished <- voice.SpeakUtteranceAndWait(ctx, utterance) }()
server.next_utterance(t)
switch operation {
case "remote":
err = voice.Cancel(t.Context())
case "caller":
cancel()
case "configuration":
// A committed replacement is modeled without cancelling local
// speech: retirement must target only the old isolated handle.
speaker.retire_remote_voice(speaker.settings)
}
if err != nil {
t.Fatal(err)
}
if err := remote_render_result(t, finished); !errors.Is(err, context.Canceled) {
t.Fatal("rendering survived cancellation", err)
}
for _, text := range []string{"local status", "local echo"} {
if id, _ := server.pending(text); id == 0 {
t.Fatal("Remote discarded local output", text)
}
}
select {
case <-server.queued:
t.Fatal("cancelled assembly exposed its suffix")
default:
}
if voice.ctx.Err() == nil {
t.Fatal("partially rendered handle was reused")
}
if _, err := speaker.open_remote_voice(t.Context()); err != nil {
t.Fatal("retired renderer retained ownership", err)
}
})
}
}
func TestRemoteSpeechRenderingRejectsWholeInvalidAssembly(t *testing.T) {
valid := remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "prefix"})
for _, frames := range [][]tentaclewire.RemoteSpeech{
nil,
{valid, {Kind: "cancel"}},
{valid, remote_render_frame(1, tentaclewire.RemoteSpeechPart{Text: "changed priority"})},
{valid, remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "bad", Language: "fr\nQUIT"})},
{valid, remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: strings.Repeat("x", 6200)})},
append(make([]tentaclewire.RemoteSpeech, 64), valid),
} {
if _, err := prepare_remote_utterance(frames); err == nil {
t.Fatal("invalid assembly admitted")
}
}
// Each frame fits IPC, but the retained assembly exceeds the byte budget.
frames := make([]tentaclewire.RemoteSpeech, 64)
for index := range frames {
frames[index] = remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: strings.Repeat("x", 5000)})
}
if _, err := prepare_remote_utterance(frames); err == nil {
t.Fatal("oversized assembly admitted")
}
for _, invalid := range []string{"bad\x00text", "bad\x1btext", "bad\x7ftext"} {
speaker, server := isolated_voice_fixture(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{valid, remote_render_frame(0,
tentaclewire.RemoteSpeechPart{End: true}, tentaclewire.RemoteSpeechPart{Text: invalid})})
if err != nil {
t.Fatal("wire-valid fixture rejected too early", err)
}
if err := voice.SpeakUtteranceAndWait(t.Context(), utterance); err == nil {
t.Fatal("unsafe text rendered")
}
select {
case <-server.queued:
t.Fatal("invalid later fragment exposed an earlier prefix")
default:
}
}
}
func TestRemoteSpeechRenderingBackendRejectionAndEmptyBoundaries(t *testing.T) {
speaker, server := isolated_voice_fixture(t)
voice, err := speaker.open_remote_voice(t.Context())
if err != nil {
t.Fatal(err)
}
utterance, err := prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0,
tentaclewire.RemoteSpeechPart{End: true}, tentaclewire.RemoteSpeechPart{End: true})})
if err != nil {
t.Fatal(err)
}
if err := voice.SpeakUtteranceAndWait(t.Context(), utterance); err != nil {
t.Fatal(err)
}
utterance, err = prepare_remote_utterance([]tentaclewire.RemoteSpeech{remote_render_frame(0,
tentaclewire.RemoteSpeechPart{Text: "never queued", Language: "rejected"})})
if err != nil {
t.Fatal(err)
}
if err := voice.SpeakUtteranceAndWait(t.Context(), utterance); err == nil {
t.Fatal("rejected metadata ignored")
}
if voice.ctx.Err() == nil {
t.Fatal("failed metadata handle survived")
}
select {
case <-server.queued:
t.Fatal("empty boundary or rejected metadata synthesized text")
default:
}
}
+131
View File
@@ -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"
"errors"
"sync"
"time"
"git.stormux.org/storm/cthulhu/internal/config"
"git.stormux.org/storm/cthulhu/internal/speech"
)
// remoteSpeechVoice owns a separate SSIP speaking/control pair. Its cancellation
// cannot target local speech, echo or status. This is a backend prerequisite,
// not a Remote lease authorizer or a production mailbox owner.
type remoteSpeechVoice struct {
client *speech.Client
ctx context.Context
cancel context.CancelFunc
once sync.Once
closed chan struct{}
closeErr error
render sync.Mutex // One typed assembly, including metadata I/O.
request sync.Mutex
renderCancel context.CancelFunc // Protected by request; never backend I/O.
}
// open_remote_voice is called only by an output worker, never an input callback.
// Acquisition neither changes the local configuration nor opens a serial device.
// There is at most one handle; a replacement requires its predecessor to close.
// ctx bounds acquisition only. A future lease owner must Close on revocation.
func (speaker *configuredSpeaker) open_remote_voice(ctx context.Context) (*remoteSpeechVoice, error) {
if !speaker.update.TryLock() {
return nil, errors.New("speech configuration is busy")
}
defer speaker.update.Unlock()
if err := ctx.Err(); err != nil {
return nil, err
}
speaker.mu.RLock()
_, software := speaker.active.(*echoSpeaker)
closed, previous := speaker.closed, speaker.remote
settings := speechSettingsFromConfig(speaker.settings)
speaker.mu.RUnlock()
if closed || !software {
return nil, errors.New("isolated Remote speech requires a connected software backend")
}
if previous != nil {
select {
case <-previous.closed:
default:
return nil, errors.New("Remote speech voice is already owned")
}
}
client := speech.NewWithSettings(speaker.socket, settings)
if err := client.Connect(ctx); err != nil {
return nil, err
}
if err := ctx.Err(); err != nil {
_ = client.Close(context.Background())
return nil, err
}
lifetime, cancel := context.WithCancel(context.Background())
voice := &remoteSpeechVoice{client: client, ctx: lifetime, cancel: cancel, closed: make(chan struct{})}
speaker.mu.Lock()
speaker.remote = voice
speaker.mu.Unlock()
return voice, nil
}
// A successful configuration commit retires the old remote voice. A failed
// preparation must leave it usable. Handles never silently adopt new settings.
func (speaker *configuredSpeaker) retire_remote_voice(settings config.Settings) {
speaker.mu.Lock()
speaker.settings = settings
remote := speaker.remote
speaker.mu.Unlock()
if remote != nil {
cleanup, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
// The local configuration has committed; cleanup cannot undo that commit.
_ = remote.Close(cleanup)
}
}
// SpeakAndWait retains ownership through the backend's END/CANCEL notification.
// Both caller cancellation and configuration retirement cancel only this client.
// This plain-text primitive deliberately does not interpret relay metadata.
func (voice *remoteSpeechVoice) SpeakAndWait(ctx context.Context, text string, began func()) error {
if err := voice.ctx.Err(); err != nil {
return err
}
request, cancel := context.WithCancel(voice.ctx)
stop := context.AfterFunc(ctx, cancel)
defer func() { stop(); cancel() }()
if err := ctx.Err(); err != nil {
return err
}
return voice.client.SpeakAndWait(request, text, began)
}
func (voice *remoteSpeechVoice) Cancel(ctx context.Context) error {
voice.request.Lock()
cancel := voice.renderCancel
if cancel != nil {
cancel()
}
voice.request.Unlock()
if cancel != nil {
// Typed cancellation retires the whole assembly even while it is
// preparing metadata and has no SSIP message to cancel yet.
return nil
}
if voice.ctx.Err() != nil {
return nil
}
return voice.client.Cancel(ctx)
}
func (voice *remoteSpeechVoice) Close(ctx context.Context) error {
voice.once.Do(func() {
voice.cancel()
voice.closeErr = errors.Join(voice.client.Cancel(ctx), voice.client.Close(ctx))
close(voice.closed)
})
return voice.closeErr
}
+433
View File
@@ -0,0 +1,433 @@
// 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
notifications bool
language, spelling, ssml string
voiceType, synthesis string
}
type isolatedVoiceUtterance struct {
id int
text, language, spelling, ssml string
module, voiceType, synthesis 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 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}
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")
}
}
+173
View File
@@ -0,0 +1,173 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package speech
import (
"context"
"errors"
"fmt"
"strings"
"unicode/utf8"
)
// Fragment is literal text with resolved spelling/language state. A fragment
// boundary is a synthesis boundary; callers coalesce transport fragments first.
// Empty language selects this client's effective configured default language.
type Fragment struct {
Text string
Characters bool
Language string
}
// SpeakFragmentsAndWait is for an isolated output owner. Unlike ordinary local
// speech, it preserves spaces/tabs and line boundaries (canonicalized to CRLF).
// Plain-text mode prevents network text being interpreted as synthesis markup.
// It validates the entire sequence before any backend operation, then retains
// the speaking lock through every END/CANCEL. Cancel still uses its own channel.
// This method implements no priority, lease, pause or local-focus policy.
func (client *Client) SpeakFragmentsAndWait(ctx context.Context, fragments []Fragment) (result error) {
if len(fragments) == 0 || len(fragments) > 8192 {
return errors.New("invalid speech fragment count")
}
bytes := 0
for _, part := range fragments {
bytes += len(part.Text)
if bytes > 256*1024 || !valid_fragment(part) {
return errors.New("invalid speech fragment")
}
}
client.mutex.Lock()
defer client.mutex.Unlock()
if client.conn == nil {
return errors.New("speech client is not connected")
}
defer func() {
if result != nil {
result = errors.Join(ctx.Err(), result)
}
if result != nil && client.conn != nil {
_ = client.conn.Close()
client.conn, client.reader = nil, nil
}
}()
// Query the effective default after initialization: a selected synthesis
// voice may override SET LANGUAGE, and the server default may be unsaved.
defaultLanguage, err := client.fragment_parameter(ctx, "LANGUAGE")
if err != nil || !valid_fragment_language(defaultLanguage, false) {
return errors.Join(ctx.Err(), errors.New("speech default language unavailable"))
}
module, err := client.fragment_parameter(ctx, "OUTPUT_MODULE")
if err != nil || !valid_fragment_language(module, false) {
return errors.Join(ctx.Err(), errors.New("speech default module unavailable"))
}
voiceType, err := client.fragment_parameter(ctx, "VOICE_TYPE")
if err != nil {
return errors.Join(ctx.Err(), errors.New("speech default voice unavailable"))
}
switch voiceType {
case "MALE1", "MALE2", "MALE3", "FEMALE1", "FEMALE2", "FEMALE3", "CHILD_MALE", "CHILD_FEMALE":
case "NO_VOICE":
if client.settings.SynthesisVoice == "" {
return errors.New("speech default voice unavailable")
}
// A named voice has no portable gender/category. Explicit remote
// languages use the SSIP baseline category instead of that voice name.
voiceType = "MALE1"
default:
return errors.New("speech default voice unavailable")
}
setVoice := func(language, named string) error {
// SET LANGUAGE can select a different output module via server config.
// Restore the reader's actual module, then clear any previous named
// voice through VOICE_TYPE. A default-language part regains its name.
for _, fields := range [][]string{
{"SET", "SELF", "LANGUAGE", language},
{"SET", "SELF", "OUTPUT_MODULE", module},
{"SET", "SELF", "VOICE_TYPE", voiceType},
} {
if err := client.sendCommand(ctx, fields); err != nil {
return err
}
}
if named != "" {
return client.sendCommand(ctx, []string{"SET", "SELF", "SYNTHESIS_VOICE", named})
}
return nil
}
if err := client.sendCommand(ctx, []string{"SET", "SELF", "SSML_MODE", "off"}); err != nil {
return err
}
for _, part := range fragments {
language, spelling, named := part.Language, "off", ""
if language == "" {
language, named = defaultLanguage, client.settings.SynthesisVoice
}
if err := setVoice(language, named); err != nil {
return err
}
if part.Characters {
spelling = "on"
}
if err := client.sendCommand(ctx, []string{"SET", "SELF", "SPELLING", spelling}); err != nil {
return err
}
text := strings.ReplaceAll(part.Text, "\r\n", "\n")
text = strings.ReplaceAll(text, "\r", "\n")
text = strings.ReplaceAll(text, "\n", "\r\n")
if err := client.speak_and_wait_locked(ctx, func() error { return client.speak_block(ctx, text) }, nil); err != nil {
return err
}
}
if err := client.sendCommand(ctx, []string{"SET", "SELF", "SPELLING", "off"}); err != nil {
return err
}
return setVoice(defaultLanguage, client.settings.SynthesisVoice)
}
// These names are fixed by the caller; response values never become commands
// until their bounded token/enum validators have accepted them.
func (client *Client) fragment_parameter(ctx context.Context, name string) (string, error) {
finish, err := prepareIO(ctx, client.conn, client.ioTimeout)
if err != nil {
return "", err
}
defer finish()
_, err = fmt.Fprint(client.conn, "GET "+name+"\r\n")
var response reply
if err == nil {
response, err = client.readReply()
}
if err != nil || response.code != 251 || len(response.data) != 1 {
return "", errors.New("speech default unavailable")
}
return response.data[0], nil
}
func valid_fragment(part Fragment) bool {
if part.Text == "" || !utf8.ValidString(part.Text) || !valid_fragment_language(part.Language, true) {
return false
}
for _, character := range part.Text {
if character < 32 && character != '\r' && character != '\n' && character != '\t' || character == 127 {
return false
}
}
return true
}
func valid_fragment_language(language string, empty bool) bool {
if language == "" {
return empty
}
if len(language) > 64 {
return false
}
for _, character := range language {
if !(character >= 'a' && character <= 'z' || character >= 'A' && character <= 'Z' || character >= '0' && character <= '9' || character == '-' || character == '_') {
return false
}
}
return true
}
+6 -1
View File
@@ -16,6 +16,11 @@ import (
func (client *Client) SpeakAndWait(ctx context.Context, text string, began func()) (result error) {
client.mutex.Lock()
defer client.mutex.Unlock()
return client.speak_and_wait_locked(ctx, func() error { return client.speak(ctx, text) }, began)
}
// The caller owns the speaking connection until the terminal notification.
func (client *Client) speak_and_wait_locked(ctx context.Context, speak func() error, began func()) (result error) {
if client.conn == nil {
return errors.New("speech client is not connected")
}
@@ -33,7 +38,7 @@ func (client *Client) SpeakAndWait(ctx context.Context, text string, began func(
}
}
}()
if err := client.speak(ctx, text); err != nil {
if err := speak(); err != nil {
return err
}
conn := client.conn
+7
View File
@@ -293,6 +293,12 @@ func (client *Client) speak(ctx context.Context, text string) (result error) {
if err != nil {
return err
}
return client.speak_block(ctx, text)
}
// text is validated by the caller. Dot escaping applies to every data line,
// including lines containing what would otherwise be the SSIP terminator.
func (client *Client) speak_block(ctx context.Context, text string) (result error) {
if client.conn == nil {
return errors.New("speech client is not connected")
@@ -315,6 +321,7 @@ func (client *Client) speak(ctx context.Context, text string) (result error) {
if strings.HasPrefix(text, ".") {
text = "." + text
}
text = strings.ReplaceAll(text, "\r\n.", "\r\n..")
request := "SPEAK\r\n" + text + "\r\n.\r\n"
if _, err := client.conn.Write([]byte(request)); err != nil {
return fmt.Errorf("send speech block: %w", err)