cthulhu/internal/app/remote_speech_stream.go

// 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"
	"log/slog"
	"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
	context   context.Context
	cleanup   func()
	current   func() bool
}

type remoteSpeechPermit struct {
	context context.Context
	current func() bool // Synchronous generation check, before cancellation callbacks run.
	release func()
}

type remoteSpeechAdmission func(tentacle.RemoteSpeechDelivery) (remoteSpeechPermit, bool)

type remoteSpeechStreamError struct {
	lease tentaclewire.RemoteLease
	cause error
}

func (err *remoteSpeechStreamError) Error() string { return err.cause.Error() }
func (err *remoteSpeechStreamError) Unwrap() error { return err.cause }

// run_remote_speech_playback supplies the isolated backend to the stream owner.
// The caller supplies original-lease policy and optional local-output admission.
func run_remote_speech_playback(ctx context.Context, speaker *configuredSpeaker,
	incoming <-chan tentacle.RemoteSpeechDelivery, allow func(tentacle.RemoteSpeechDelivery) bool,
	admit remoteSpeechAdmission, logger *slog.Logger) 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, admit, 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")
		}
		if logger != nil && logger.Enabled(request, slog.LevelDebug) {
			for _, fragment := range utterance.fragments {
				logger.DebugContext(request, "delivering remote speech", "utterance", fragment.Text,
					"characters", fragment.Characters, "language", fragment.Language, "priority", utterance.priority)
			}
		}
		err := voice.SpeakUtteranceAndWait(request, utterance)
		if err != nil && request.Err() == nil && logger != nil {
			logger.Debug("remote speech rendering failed", "error", err)
		}
		return err
	})
}

// run_remote_speech_stream retains complete deliveries until cleanup. 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,
	admit remoteSpeechAdmission,
	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{}
	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) {
		item.cleanup()
		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) || item.context.Err() != nil || !item.current() {
				release(item)
				continue
			}
			request, stop := context.WithCancel(item.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 || !active.current()
			failedLease := active.delivery.Lease
			cancel()
			release(active)
			active, cancel, finished = nil, nil, nil
			if err != nil && !interrupted && !errors.Is(err, context.Canceled) {
				return &remoteSpeechStreamError{lease: failedLease, cause: 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 = delivery.Lease, delivery.Context.Done()
			}
			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()
					}
					// Pause cancels the current assembly because this backend
					// cannot resume it. It must not mute future utterances: NVDA
					// clears pause on cancel or fresh speech, without requiring
					// a separate pause(false) message on the relay.
					delivery.Release()
					continue
				}
			}
			utterance, err := prepare_remote_utterance(delivery.Speeches)
			if err != nil {
				delivery.Release()
				return &remoteSpeechStreamError{lease: delivery.Lease, cause: err}
			}
			permit := remoteSpeechPermit{context: delivery.Context, release: func() {}, current: func() bool { return true }}
			if admit != nil {
				admitted, ok := admit(delivery)
				if !ok {
					delivery.Release()
					continue
				}
				permit = admitted
			}
			item := &remoteSpeechStreamItem{delivery: delivery, utterance: utterance, context: permit.context, cleanup: permit.release, current: permit.current}
			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 {
				permit.release()
				delivery.Release()
				return &remoteSpeechStreamError{lease: delivery.Lease, cause: 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
		}
	}
}