cthulhu/internal/app/tentacle_remote_speech_test.go

// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later

package app

import (
	"context"
	"log/slog"
	"testing"
	"time"

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

func TestRemoteSpeechLocalPriorityDiscardsActiveAndQueuedWork(t *testing.T) {
	for _, change := range []string{"announcement", "interrupt", "mode", "disable", "secure"} {
		t.Run(change, func(t *testing.T) {
			worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
			worker.running, worker.workerContext = true, t.Context()
			f := remote_stream_fixture(t, false, worker.admit_remote_speech)
			activeRelease := f.speak(t, "old active", 2)
			active := f.next(t, "old active")
			pendingRelease := f.speak(t, "old pending", 2)
			// Equal priority keeps the queued item behind this active NOW.
			// Use the barrier to establish admission before local output.
			f.barrier(t)
			switch change {
			case "announcement":
				worker.Enqueue(speechRequest{Text: "local status"})
			case "interrupt":
				worker.InterruptForUser()
			case "mode":
				worker.cycle_speech_mode()
			case "disable":
				worker.SetEnabled(false)
			case "secure":
				worker.SetSecureInputActive(true)
			}
			remote_stream_wait(t, active.stopped)
			remote_stream_wait(t, activeRelease)
			remote_stream_wait(t, pendingRelease)
			remote_stream_wait(t, f.speak(t, "blocked by local policy", 2))
			select {
			case <-f.started:
				t.Fatal("old or blocked incoming speech reached rendering")
			default:
			}
			worker.mutex.Lock()
			worker.pending, worker.interrupting = nil, false
			worker.enabled, worker.secure, worker.mode = true, false, speechModeSpeech
			worker.mutex.Unlock()
			freshRelease := f.speak(t, "fresh output", 0)
			fresh := f.next(t, "fresh output")
			fresh.finish <- nil
			remote_stream_wait(t, freshRelease)
			f.stop(t)
		})
	}
}

func TestRemoteSpeechLocalPriorityRejectsOldMailboxAndIncompleteAssembly(t *testing.T) {
	worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
	worker.running, worker.workerContext = true, t.Context()
	// The first fragment's timestamp survives finalization: work retained in
	// the IPC assembly or mailbox must not acquire fresh local authority later.
	old := tentacle.RemoteSpeechDelivery{Context: t.Context(), Received: time.Now()}
	worker.Enqueue(speechRequest{Text: "local status"})
	worker.mutex.Lock()
	worker.pending = nil // Local queue drained before the old mailbox is read.
	worker.mutex.Unlock()
	if _, ok := worker.admit_remote_speech(old); ok {
		t.Fatal("old mailbox/assembly work followed local output")
	}
	fresh := tentacle.RemoteSpeechDelivery{Context: t.Context(), Received: time.Now()}
	permit, ok := worker.admit_remote_speech(fresh)
	if !ok || permit.context.Err() != nil || !permit.current() {
		t.Fatal("fresh incoming output was rejected after local reading")
	}
	permit.release()
}

func TestRemoteSpeechRejectedLocalRequestPreservesRemoteAndServerCancelContinues(t *testing.T) {
	worker := newSpeechWorker(slog.New(slog.DiscardHandler), nil)
	worker.running, worker.workerContext = true, t.Context()
	f := remote_stream_fixture(t, false, worker.admit_remote_speech)
	released := f.speak(t, "active", 0)
	active := f.next(t, "active")
	oldGeneration := worker.generation + 1
	worker.Enqueue(speechRequest{Text: "stale result", presentationGeneration: &oldGeneration})
	select {
	case <-active.ctx.Done():
		t.Fatal("rejected local request cancelled Remote output")
	default:
	}
	// A server priority cancellation discards this assembly. It must not
	// disable the owner or retire consent, and no suffix is retried.
	active.finish <- context.Canceled
	remote_stream_wait(t, released)
	nextRelease := f.speak(t, "after cancellation", 0)
	next := f.next(t, "after cancellation")
	next.finish <- nil
	remote_stream_wait(t, nextRelease)
	f.stop(t)
}

func TestRemoteSpeechBackgroundPriorityRejectionClosesCandidate(t *testing.T) {
	speaker, server := isolated_voice_fixture(t)
	server.mutex.Lock()
	server.queryReplies = map[string]string{"SET SELF PRIORITY TEXT": "400 Rejected\r\n"}
	server.mutex.Unlock()
	if _, err := speaker.open_remote_voice(t.Context()); err == nil {
		t.Fatal("isolated client acquired without lower backend priority")
	}
	speaker.mu.RLock()
	remote := speaker.remote
	speaker.mu.RUnlock()
	if remote != nil {
		t.Fatal("failed candidate became the owned voice")
	}
	server.mutex.Lock()
	server.queryReplies = nil
	server.mutex.Unlock()
	voice, err := speaker.open_remote_voice(t.Context())
	if err != nil {
		t.Fatal("failed priority setup prevented a fresh acquisition", err)
	}
	if err := voice.Close(t.Context()); err != nil {
		t.Fatal(err)
	}
}