cthulhu/internal/app/speech_worker.go

// Cthulhu screen reader.
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along with
// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.

package app

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"git.stormux.org/storm/cthulhu/internal/atspi"
	"git.stormux.org/storm/cthulhu/internal/config"
	"git.stormux.org/storm/cthulhu/internal/sound"
)

// speechQueueCapacity bounds work retained while the speech backend is slow.
// Requests remain FIFO. On overflow, the oldest pending request is discarded
// so that current focus is favored, and the loss is reported without content.
const speechQueueCapacity = 64

type speechRequest struct {
	presentationGeneration *uint64 // Checked under the speech owner mutex.
	presentationRevision   *uint64

	Window             atspi.ObjectID
	IntroducesWindow   bool
	ReadAll            []readAllPart
	CustomEcho         bool
	Character          bool
	VoiceType          string
	Line               *speechLine
	Context            []atspi.Snapshot
	Spelling           []string // One cancellable queue item, with separate letter utterances.
	SpellingVoices     []string
	Sound              []sound.Cue
	Chunks             []string // Presentation pause boundaries; one cancellable queue item.
	FallbackChunks     []string
	Text               string
	FallbackText       string
	Role               string
	Cause              string
	Append             bool
	ClipboardTimestamp uint32
	BypassFocusGate    bool
	ExcludeFromHistory bool
	ForceSpeech        bool // Only the speech-mode announcement bypasses the mode.
	Confirmation       bool // Wait for speech completion before automatic output.
	Uninterruptible    bool // Accepted tentacle message; ordinary input cannot cancel it.
	confirmationID     uint64
	deliveryGeneration uint64
	context            context.Context
	cancel             context.CancelFunc
	result             chan error
}

type readAllPart struct {
	Text      string
	VoiceType string
	Line      *speechLine
	Context   []atspi.Snapshot
	Began     func()
}

// speechWorker owns the pending queue under mutex. Delivery and cancellation
// have separate goroutines and joins; activeDone and generation form their
// barrier so cancellation cannot overtake newly accepted output.
type speechWorker struct {
	logger         *slog.Logger
	speaker        Speaker
	history        *speechHistory
	beep           func(context.Context) error
	textTransform  func(string) string
	reading        func() config.ReadingSettings
	verbosity      func() int
	indentation    indentationHistory // Owned by the serialized delivery goroutine.
	speechContext  []atspi.Snapshot   // Same owner; replaced on each known navigation context.
	indentationCue func(sound.Cue)

	mutex              sync.Mutex
	enabled            bool
	mode               speechMode
	beepCancel         context.CancelFunc
	pending            []speechRequest
	wake               chan struct{}
	interrupt          chan struct{}
	runDone            chan struct{}
	cancelDone         chan struct{}
	cancel             context.CancelFunc
	secure             bool
	started            bool
	running            bool
	workerContext      context.Context
	overflow           uint64
	reported           uint64
	generation         uint64
	interrupting       bool
	activeDone         chan struct{}
	startupUntil       time.Time
	readAllCancel      context.CancelFunc
	readAllTimestamp   uint32
	commandTimestamp   uint32
	lastInputTimestamp uint32
	titleWindow        atspi.ObjectID
	titleInput         titleInputPolicy
	confirmationID     uint64 // Owned under mutex; reset on completion or user interruption.
	confirmationSerial uint64

	presentationDelivery   context.Context // Replaced when the next request reaches delivery.
	presentationRevision   uint64
	uninterruptibleContext context.Context
	uninterruptibleCancel  context.CancelFunc
}

func newSpeechWorker(logger *slog.Logger, speaker Speaker) *speechWorker {
	return &speechWorker{
		logger:    logger,
		speaker:   speaker,
		enabled:   true,
		wake:      make(chan struct{}, 1),
		interrupt: make(chan struct{}, 1),
	}
}

func (worker *speechWorker) Start(ctx context.Context) error {
	worker.mutex.Lock()
	defer worker.mutex.Unlock()
	if worker.started {
		return errors.New("speech worker already started")
	}
	workerContext, cancel := context.WithCancel(ctx)
	worker.workerContext = workerContext
	worker.cancel = cancel
	worker.started = true
	worker.running = true
	worker.runDone = make(chan struct{})
	worker.cancelDone = make(chan struct{})
	go worker.run(workerContext)
	go worker.cancelActiveSpeech(workerContext)
	return nil
}

func (worker *speechWorker) Enqueue(request speechRequest) {
	worker.mutex.Lock()
	if request.presentationGeneration != nil && (*request.presentationGeneration != worker.generation ||
		request.presentationRevision == nil || *request.presentationRevision != worker.presentationRevision) {
		worker.mutex.Unlock()
		completeSpeechRequest(request, context.Canceled)
		return
	}
	if !worker.enabled {
		worker.mutex.Unlock()
		completeSpeechRequest(request, context.Canceled)
		return
	}
	// Equal millisecond timestamps can belong to the copy key and its result.
	if request.ClipboardTimestamp != 0 && worker.lastInputTimestamp != 0 && int32(request.ClipboardTimestamp-worker.lastInputTimestamp) < 0 {
		worker.mutex.Unlock()
		completeSpeechRequest(request, context.Canceled)
		return
	}
	if worker.secure && !strings.HasPrefix(request.Cause, "remote-") {
		worker.mutex.Unlock()
		completeSpeechRequest(request, context.Canceled)
		return
	}
	if worker.readAllCancel != nil && request.Cause != "read-all" && !request.Append {
		worker.readAllCancel()
		worker.readAllCancel = nil
	}
	if request.context != nil && request.context.Err() != nil {
		worker.mutex.Unlock()
		completeSpeechRequest(request, request.context.Err())
		return
	}
	if !worker.running {
		worker.mutex.Unlock()
		completeSpeechRequest(request, errors.New("speech worker is not running"))
		return
	}
	if request.Confirmation {
		worker.confirmationSerial++
		request.confirmationID = worker.confirmationSerial
		worker.confirmationID = request.confirmationID
	} else if worker.confirmationID != 0 {
		request.Append = true
	}
	if worker.has_uninterruptible_locked() {
		request.Append = true
	}
	// Pair a title with the next announcement from its window. User input
	// and explicit interruption retire this pairing; another window starts
	// normally. This queues the continuation without protecting the title
	// against navigation or estimating how long it will take to speak.
	if worker.titleWindow.Valid() && request.Window == worker.titleWindow && !request.IntroducesWindow {
		request.Append = true
	}
	worker.titleWindow = atspi.ObjectID{}
	if request.IntroducesWindow && request.Text != "" {
		worker.titleWindow = request.Window
	}
	if time.Now().Before(worker.startupUntil) && request.Cause != "startup" {
		request.Append = true
	}
	if len(worker.pending) == speechQueueCapacity {
		if worker.pending[0].confirmationID == worker.confirmationID {
			worker.confirmationID = 0
		}
		completeSpeechRequest(worker.pending[0], errors.New("speech request dropped after queue overflow"))
		copy(worker.pending, worker.pending[1:])
		worker.pending = worker.pending[:speechQueueCapacity-1]
		worker.overflow++
	}
	worker.titleInput.observe(request)
	worker.presentationRevision++
	worker.pending = append(worker.pending, request)
	worker.mutex.Unlock()

	select {
	case worker.wake <- struct{}{}:
	default:
	}
}

func (worker *speechWorker) EnqueueAndWait(ctx context.Context, request speechRequest) error {
	worker.mutex.Lock()
	if worker.running {
		requestContext, cancelRequest := context.WithCancel(worker.workerContext)
		request.context = requestContext
		request.cancel = cancelRequest
	}
	worker.mutex.Unlock()

	request.result = make(chan error, 1)
	worker.Enqueue(request)
	stopCallerWatch := context.AfterFunc(ctx, func() {
		if request.cancel != nil {
			request.cancel()
		}
	})
	select {
	case err := <-request.result:
		if request.cancel != nil {
			request.cancel()
		}
		stopCallerWatch()
		return err
	case <-ctx.Done():
		if request.cancel != nil {
			request.cancel()
		}
		stopCallerWatch()
		return ctx.Err()
	}
}

func (worker *speechWorker) Stop(ctx context.Context) error {
	worker.mutex.Lock()
	if !worker.started {
		worker.mutex.Unlock()
		return nil
	}
	worker.running = false
	worker.confirmationID = 0
	if worker.uninterruptibleCancel != nil {
		worker.uninterruptibleCancel()
	}
	if worker.readAllCancel != nil {
		worker.readAllCancel()
		worker.readAllCancel = nil
	}
	completeSpeechRequests(worker.pending, context.Canceled)
	worker.pending = nil
	cancel := worker.cancel
	runDone := worker.runDone
	cancelDone := worker.cancelDone
	worker.mutex.Unlock()

	cancel()
	for _, done := range []chan struct{}{runDone, cancelDone} {
		select {
		case <-done:
		case <-ctx.Done():
			worker.reportOverflow()
			return fmt.Errorf("wait for speech worker: %w", ctx.Err())
		}
	}
	worker.reportOverflow()
	return nil
}

func (worker *speechWorker) run(ctx context.Context) {
	defer close(worker.runDone)
	defer func() {
		worker.mutex.Lock()
		worker.pending = nil
		worker.running = false
		worker.mutex.Unlock()
	}()

	for {
		select {
		case <-ctx.Done():
			return
		case <-worker.wake:
		}
		worker.reportOverflow()

		for {
			if ctx.Err() != nil {
				return
			}
			request, ok := worker.next()
			if !ok {
				break
			}
			deliveryContext := ctx
			if request.context != nil {
				deliveryContext = request.context
			}
			err := worker.deliver(deliveryContext, request)
			completeSpeechRequest(request, err)
			worker.mutex.Lock()
			worker.uninterruptibleContext, worker.uninterruptibleCancel = nil, nil
			if request.confirmationID == worker.confirmationID {
				worker.confirmationID = 0
			}
			close(worker.activeDone)
			worker.activeDone = nil
			worker.mutex.Unlock()
			if err != nil {
				if ctx.Err() != nil {
					return
				}
				worker.logger.Error("deliver accessible presentation", "role", request.Role, "error", err)
			}
		}
	}
}

func completeSpeechRequests(requests []speechRequest, err error) {
	for _, request := range requests {
		completeSpeechRequest(request, err)
	}
}

func completeSpeechRequest(request speechRequest, err error) {
	if request.Uninterruptible && request.cancel != nil {
		request.cancel()
	}
	if request.result == nil {
		return
	}
	select {
	case request.result <- err:
	default:
	}
}

func (worker *speechWorker) reportOverflow() {
	worker.mutex.Lock()
	if worker.overflow == worker.reported {
		worker.mutex.Unlock()
		return
	}
	dropped := worker.overflow - worker.reported
	worker.reported = worker.overflow
	total := worker.overflow
	worker.mutex.Unlock()
	worker.logger.Warn(
		"speech queue overflow",
		"dropped_requests", dropped,
		"total_dropped_requests", total,
	)
}

func (worker *speechWorker) next() (speechRequest, bool) {
	worker.mutex.Lock()
	defer worker.mutex.Unlock()
	if !worker.running || worker.interrupting || len(worker.pending) == 0 {
		return speechRequest{}, false
	}
	worker.activeDone = make(chan struct{})
	request := worker.pending[0]
	if request.Uninterruptible {
		worker.uninterruptibleContext, worker.uninterruptibleCancel = request.context, request.cancel
	}
	worker.presentationDelivery = nil
	if request.presentationGeneration != nil {
		worker.presentationDelivery = request.context
	}
	request.deliveryGeneration = worker.generation
	copy(worker.pending, worker.pending[1:])
	worker.pending[len(worker.pending)-1] = speechRequest{}
	worker.pending = worker.pending[:len(worker.pending)-1]
	return request, true
}