cthulhu/internal/app/remote_speech_stream_test.go

// 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"
	"sync/atomic"
	"testing"
	"time"

	"git.stormux.org/storm/cthulhu/internal/tentacle"
	"git.stormux.org/storm/cthulhu/internal/tentaclewire"
)

type remoteStreamCall struct {
	ctx       context.Context
	utterance remoteSpeechUtterance
	finish    chan error
	stopped   chan struct{}
	cleanup   chan struct{}
}

type remoteStreamFixture struct {
	incoming chan tentacle.RemoteSpeechDelivery
	started  chan remoteStreamCall
	done     chan error
	ctx      context.Context
	cancel   context.CancelFunc
	lease    tentaclewire.RemoteLease
	hold     bool
	allowed  atomic.Bool
}

func remote_stream_fixture(t *testing.T, hold bool, admission ...remoteSpeechAdmission) *remoteStreamFixture {
	t.Helper()
	ctx, cancel := context.WithCancel(t.Context())
	f := &remoteStreamFixture{ctx: ctx, cancel: cancel, hold: hold,
		incoming: make(chan tentacle.RemoteSpeechDelivery, 128), started: make(chan remoteStreamCall, 128), done: make(chan error, 1),
		lease: tentaclewire.RemoteLease{Lease: strings.Repeat("a", 32), Epoch: 1}}
	f.allowed.Store(true)
	var admit remoteSpeechAdmission
	if len(admission) != 0 {
		admit = admission[0]
	}
	go func() {
		f.done <- run_remote_speech_stream(ctx, f.incoming, func(tentacle.RemoteSpeechDelivery) bool { return f.allowed.Load() }, admit,
			func(ctx context.Context, _ tentacle.RemoteSpeechDelivery, utterance remoteSpeechUtterance) error {
				call := remoteStreamCall{ctx: ctx, utterance: utterance, finish: make(chan error, 1), stopped: make(chan struct{}), cleanup: make(chan struct{})}
				f.started <- call
				select {
				case err := <-call.finish:
					return err
				case <-ctx.Done():
					close(call.stopped)
					if hold {
						<-call.cleanup
					}
					return ctx.Err()
				}
			})
	}()
	t.Cleanup(cancel)
	return f
}

func remote_stream_wait(t *testing.T, done <-chan struct{}) {
	t.Helper()
	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("Remote stream did not reach its synchronization point")
	}
}

func (f *remoteStreamFixture) send(t *testing.T, effect tentaclewire.RemoteSpeech) <-chan struct{} {
	t.Helper()
	return f.send_delivery(t, f.ctx, f.lease, []tentaclewire.RemoteSpeech{effect})
}

func (f *remoteStreamFixture) send_delivery(t *testing.T, ctx context.Context, lease tentaclewire.RemoteLease, frames []tentaclewire.RemoteSpeech) <-chan struct{} {
	t.Helper()
	released := make(chan struct{})
	var calls atomic.Int32
	f.incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease, Received: time.Now(), Speeches: frames, Release: func() {
		if calls.Add(1) != 1 {
			t.Error("Remote stream released mailbox budget twice")
			return
		}
		close(released)
	}}
	return released
}

func (f *remoteStreamFixture) speak(t *testing.T, text string, priority int) <-chan struct{} {
	return f.send(t, remote_render_frame(priority, tentaclewire.RemoteSpeechPart{Text: text}))
}

func (f *remoteStreamFixture) next(t *testing.T, text string) remoteStreamCall {
	t.Helper()
	select {
	case call := <-f.started:
		if len(call.utterance.fragments) != 1 || call.utterance.fragments[0].Text != text {
			t.Fatalf("unexpected Remote rendering: %#v", call.utterance)
		}
		return call
	case <-time.After(3 * time.Second):
		t.Fatal("Remote stream did not start rendering")
		return remoteStreamCall{}
	}
}

func (f *remoteStreamFixture) barrier(t *testing.T) {
	t.Helper()
	paused := false
	remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "pause", Paused: &paused}))
}

func (f *remoteStreamFixture) stop(t *testing.T) {
	t.Helper()
	f.cancel()
	err := remote_render_result(t, f.done)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("stream shutdown = %v", err)
	}
}

func TestRemoteSpeechStreamPriorityAndCleanup(t *testing.T) {
	f := remote_stream_fixture(t, true)
	firstRelease := f.speak(t, "active", 0)
	first := f.next(t, "active")
	normalRelease := f.speak(t, "normal", 0)
	nextRelease := f.speak(t, "next", 1)
	f.barrier(t)
	select {
	case <-first.stopped:
		t.Fatal("NEXT interrupted the current assembly")
	default:
	}
	nowRelease := f.speak(t, "now", 2)
	remote_stream_wait(t, first.stopped)
	f.barrier(t)
	select {
	case <-firstRelease:
		t.Fatal("active budget was freed before renderer cleanup")
	case <-f.started:
		t.Fatal("replacement rendering overtook cleanup")
	default:
	}
	close(first.cleanup)
	now := f.next(t, "now")
	remote_stream_wait(t, firstRelease)
	secondNowRelease := f.speak(t, "second now", 2)
	f.barrier(t)
	select {
	case <-now.stopped:
		t.Fatal("NOW interrupted equal-priority speech")
	default:
	}
	now.finish <- nil
	second := f.next(t, "second now")
	second.finish <- nil
	next := f.next(t, "next")
	next.finish <- nil
	normal := f.next(t, "normal")
	normal.finish <- nil
	for _, release := range []<-chan struct{}{normalRelease, nextRelease, nowRelease, secondNowRelease} {
		remote_stream_wait(t, release)
	}
	f.stop(t)
}

func TestRemoteSpeechStreamCancelAndPause(t *testing.T) {
	for _, kind := range []string{"cancel", "pause"} {
		t.Run(kind, func(t *testing.T) {
			f := remote_stream_fixture(t, true)
			activeRelease := f.speak(t, "active", 0)
			active := f.next(t, "active")
			pendingRelease := f.speak(t, "discard pending", 0)
			effect := tentaclewire.RemoteSpeech{Kind: kind}
			if kind == "pause" {
				paused := true
				effect.Paused = &paused
			}
			remote_stream_wait(t, f.send(t, effect))
			remote_stream_wait(t, active.stopped)
			remote_stream_wait(t, pendingRelease)
			futureRelease := f.speak(t, "future", 0)
			close(active.cleanup)
			future := f.next(t, "future")
			remote_stream_wait(t, activeRelease)
			future.finish <- nil
			remote_stream_wait(t, futureRelease)
			f.stop(t)
		})
	}
}

func TestRemoteSpeechStreamFreshSpeechAfterPause(t *testing.T) {
	for _, cancelFirst := range []bool{false, true} {
		t.Run(map[bool]string{false: "speak", true: "cancel then speak"}[cancelFirst], func(t *testing.T) {
			f := remote_stream_fixture(t, false)
			paused := true
			remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "pause", Paused: &paused}))
			if cancelFirst {
				remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "cancel"}))
			}
			released := f.speak(t, "new navigation", 0)
			call := f.next(t, "new navigation")
			call.finish <- nil
			remote_stream_wait(t, released)
			f.stop(t)
		})
	}
}

func TestRemoteSpeechStreamRevocationAndReplacement(t *testing.T) {
	f := remote_stream_fixture(t, true)
	leaseContext, revoke := context.WithCancel(f.ctx)
	firstRelease := f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "old"})})
	first := f.next(t, "old")
	pendingRelease := f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "old pending"})})
	f.barrier(t)
	revoke()
	remote_stream_wait(t, first.stopped)
	remote_stream_wait(t, pendingRelease)
	remote_stream_wait(t, f.send_delivery(t, leaseContext, f.lease, []tentaclewire.RemoteSpeech{remote_render_frame(2, tentaclewire.RemoteSpeechPart{Text: "stale"})}))
	f.lease.Epoch++
	newRelease := f.speak(t, "replacement", 0)
	f.barrier(t)
	close(first.cleanup)
	replacement := f.next(t, "replacement")
	remote_stream_wait(t, firstRelease)
	replacement.finish <- nil
	remote_stream_wait(t, newRelease)
	f.stop(t)
}

func TestRemoteSpeechStreamEOFRetainsBudgetThroughCleanup(t *testing.T) {
	f := remote_stream_fixture(t, true)
	activeRelease := f.speak(t, "active", 0)
	active := f.next(t, "active")
	pendingRelease := f.speak(t, "pending", 0)
	f.barrier(t)
	close(f.incoming)
	remote_stream_wait(t, active.stopped)
	remote_stream_wait(t, pendingRelease)
	select {
	case <-activeRelease:
		t.Fatal("EOF released active budget before cleanup")
	case <-f.done:
		t.Fatal("EOF returned before renderer cleanup")
	default:
	}
	close(active.cleanup)
	if err := remote_render_result(t, f.done); err != nil {
		t.Fatal(err)
	}
	remote_stream_wait(t, activeRelease)
}

func TestRemoteSpeechStreamRechecksPolicyAndFailsClosed(t *testing.T) {
	for _, cause := range []string{"policy", "backend", "invalid", "overflow", "bytes"} {
		t.Run(cause, func(t *testing.T) {
			f := remote_stream_fixture(t, false)
			activeRelease := f.speak(t, "active", 0)
			active := f.next(t, "active")
			pendingRelease := f.speak(t, "pending", 0)
			f.barrier(t)
			switch cause {
			case "policy":
				f.allowed.Store(false)
				active.finish <- nil
				remote_stream_wait(t, pendingRelease)
				remote_stream_wait(t, f.speak(t, "denied", 2))
				f.stop(t)
			case "backend":
				active.finish <- errors.New("backend failure")
				if err := remote_render_result(t, f.done); err == nil || errors.Is(err, context.Canceled) {
					t.Fatalf("backend failure did not retire the stream: %v", err)
				}
			case "invalid":
				remote_stream_wait(t, f.send(t, tentaclewire.RemoteSpeech{Kind: "speak"}))
				// Release precedes error return; wait for the owned result.
				if err := remote_render_result(t, f.done); err == nil {
					t.Fatal("invalid assembly was accepted")
				}
			case "overflow":
				for range tentaclewire.RemoteSpeechQueueSize - 2 {
					f.speak(t, "bounded", 0)
				}
				remote_stream_wait(t, f.speak(t, "overflow", 0))
				if err := remote_render_result(t, f.done); err == nil {
					t.Fatal("owner exceeded retained frame budget")
				}
			case "bytes":
				// Frame count remains below 64 while encoded bytes exceed 256 KiB.
				for range 52 {
					f.speak(t, strings.Repeat("x", 5000), 0)
				}
				remote_stream_wait(t, f.speak(t, strings.Repeat("x", 5000), 0))
				if err := remote_render_result(t, f.done); err == nil {
					t.Fatal("owner exceeded retained byte budget")
				}
			}
			remote_stream_wait(t, activeRelease)
			remote_stream_wait(t, pendingRelease)
		})
	}
}

func TestRemoteSpeechStreamIsolatedBackendCancellation(t *testing.T) {
	speaker, server := isolated_voice_fixture(t)
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	incoming := make(chan tentacle.RemoteSpeechDelivery, 8)
	done := make(chan error, 1)
	go func() {
		done <- run_remote_speech_playback(ctx, speaker, incoming, func(tentacle.RemoteSpeechDelivery) bool { return true }, nil, nil)
	}()
	lease := tentaclewire.RemoteLease{Lease: strings.Repeat("b", 32), Epoch: 1}
	send := func(effect tentaclewire.RemoteSpeech) <-chan struct{} {
		released := make(chan struct{})
		incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease, Speeches: []tentaclewire.RemoteSpeech{effect}, Release: func() { close(released) }}
		return released
	}
	localDone := make(chan error, 1)
	go func() { localDone <- speaker.SpeakAndWait(ctx, "local status", nil) }()
	local := server.next_utterance(t)
	remoteRelease := send(remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "remote first"}))
	remote := server.next_utterance(t)
	if remote.priority != "TEXT" || local.priority != "MESSAGE" {
		t.Fatal("incoming speech can compete with local output priority")
	}
	if local.id == remote.id {
		t.Fatal("stream acquired the local speaking client")
	}
	remote_stream_wait(t, send(tentaclewire.RemoteSpeech{Kind: "cancel"}))
	remote_stream_wait(t, remoteRelease)
	server.mutex.Lock()
	localPending := server.connections[local.id].pending
	server.mutex.Unlock()
	if localPending == "" {
		t.Fatal("remote cancel removed local status")
	}
	nextRelease := send(remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "remote after cancel"}))
	next := server.next_utterance(t)
	if next.text != "remote after cancel" || next.id == local.id || next.id == remote.id {
		t.Fatal("cancelled renderer was reused or replayed")
	}
	server.end_utterance(t, next)
	remote_stream_wait(t, nextRelease)
	server.end_utterance(t, local)
	if err := remote_render_result(t, localDone); err != nil {
		t.Fatal(err)
	}
	cancel()
	if err := remote_render_result(t, done); !errors.Is(err, context.Canceled) {
		t.Fatalf("stream shutdown: %v", err)
	}
}

func TestRemoteSpeechStreamEarlyBackendCompletionKeepsReading(t *testing.T) {
	for _, terminal := range []int{702, 703} {
		t.Run(fmt.Sprint(terminal), func(t *testing.T) {
			speaker, server := isolated_voice_fixture(t)
			server.mutex.Lock()
			server.queryReplies = map[string]string{"SPEAK": fmt.Sprintf(
				"230 OK\r\n%d-1\r\n%d-1\r\n%d DONE\r\n225-1\r\n225 QUEUED\r\n", terminal, terminal, terminal)}
			server.mutex.Unlock()
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			incoming := make(chan tentacle.RemoteSpeechDelivery, 8)
			done := make(chan error, 1)
			go func() {
				done <- run_remote_speech_playback(ctx, speaker, incoming, func(tentacle.RemoteSpeechDelivery) bool { return true }, nil, nil)
			}()
			lease := tentaclewire.RemoteLease{Lease: strings.Repeat("b", 32), Epoch: 1}
			for _, text := range []string{"first", "second", "third"} {
				released := make(chan struct{})
				incoming <- tentacle.RemoteSpeechDelivery{Context: ctx, Lease: lease,
					Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: text})},
					Release:  func() { close(released) }}
				if got := server.next_utterance(t); got.text != text {
					t.Fatalf("remote output = %q, want %q", got.text, text)
				}
				remote_stream_wait(t, released)
			}
			cancel()
			if err := remote_render_result(t, done); !errors.Is(err, context.Canceled) {
				t.Fatalf("stream shutdown: %v", err)
			}
		})
	}
}

func TestRemoteSpeechStreamPolicyLossDuringAcquisition(t *testing.T) {
	speaker, server := isolated_voice_fixture(t)
	held, resume := make(chan struct{}), make(chan struct{})
	server.mutex.Lock()
	server.holdCommand, server.held, server.resume = "HISTORY GET CLIENT_ID", held, resume
	server.mutex.Unlock()
	// Unblock the private server even if an assertion fails.
	defer func() {
		select {
		case <-resume:
		default:
			close(resume)
		}
	}()
	var allowed atomic.Bool
	allowed.Store(true)
	incoming := make(chan tentacle.RemoteSpeechDelivery, 1)
	released := make(chan struct{})
	incoming <- tentacle.RemoteSpeechDelivery{Context: t.Context(),
		Lease:    tentaclewire.RemoteLease{Lease: strings.Repeat("c", 32), Epoch: 1},
		Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: "denied after acquisition"})},
		Release:  func() { close(released) }}
	done := make(chan error, 1)
	go func() {
		done <- run_remote_speech_playback(t.Context(), speaker, incoming, func(tentacle.RemoteSpeechDelivery) bool { return allowed.Load() }, nil, nil)
	}()
	remote_stream_wait(t, held)
	allowed.Store(false)
	server.mutex.Lock()
	server.holdCommand = ""
	server.mutex.Unlock()
	close(resume)
	if err := remote_render_result(t, done); err == nil {
		t.Fatal("policy loss during acquisition was accepted")
	}
	remote_stream_wait(t, released)
	select {
	case <-server.queued:
		t.Fatal("policy loss exposed speech to the backend")
	default:
	}
}

func TestRemoteSpeechStreamVoiceRetiresWithOriginalLease(t *testing.T) {
	for _, state := range []string{"active", "idle"} {
		t.Run(state, func(t *testing.T) { test_remote_stream_lease_voice(t, state == "active") })
	}
}

func test_remote_stream_lease_voice(t *testing.T, active bool) {
	t.Helper()
	speaker, server := isolated_voice_fixture(t)
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	leaseContext, revoke := context.WithCancel(ctx)
	defer revoke()
	incoming := make(chan tentacle.RemoteSpeechDelivery, 4)
	done := make(chan error, 1)
	go func() {
		done <- run_remote_speech_playback(ctx, speaker, incoming, func(delivery tentacle.RemoteSpeechDelivery) bool { return delivery.Context.Err() == nil }, nil, nil)
	}()
	send := func(lifetime context.Context, epoch uint64, text string) <-chan struct{} {
		released := make(chan struct{})
		incoming <- tentacle.RemoteSpeechDelivery{Context: lifetime,
			Lease:    tentaclewire.RemoteLease{Lease: strings.Repeat("d", 32), Epoch: epoch},
			Speeches: []tentaclewire.RemoteSpeech{remote_render_frame(0, tentaclewire.RemoteSpeechPart{Text: text})},
			Release:  func() { close(released) }}
		return released
	}
	firstRelease := send(leaseContext, 1, "first lease")
	first := server.next_utterance(t)
	if !active {
		server.end_utterance(t, first)
		remote_stream_wait(t, firstRelease)
	}
	speaker.mu.RLock()
	oldVoice := speaker.remote
	speaker.mu.RUnlock()
	revoke()
	remote_stream_wait(t, oldVoice.closed)
	remote_stream_wait(t, firstRelease)
	nextRelease := send(ctx, 2, "replacement lease")
	next := server.next_utterance(t)
	if next.id == first.id || next.text != "replacement lease" {
		t.Fatal("replacement inherited the retired backend client")
	}
	// Repeated old cleanup cannot cancel the replacement's pending message.
	if err := oldVoice.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
	server.mutex.Lock()
	pending := server.connections[next.id].pending
	server.mutex.Unlock()
	if pending != "replacement lease" {
		t.Fatal("old lease cleanup removed replacement speech")
	}
	server.end_utterance(t, next)
	remote_stream_wait(t, nextRelease)
	cancel()
	if err := remote_render_result(t, done); !errors.Is(err, context.Canceled) {
		t.Fatalf("stream shutdown: %v", err)
	}
}