cthulhu/internal/tentacle/remote_output_test.go

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

package tentacle

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

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

func TestRemoteSpeechOutputSDKTicket(t *testing.T) {
	for _, language := range []string{"go", "python"} {
		t.Run(language, func(t *testing.T) {
			manager := test_manager(t, DefaultLimits(), nil)
			session := startSessionSDK(t, manager, language, "speech")
			if manager.CaptureRemoteSpeechOutput() != nil {
				t.Fatal("captured output without consent")
			}
			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			if manager.CaptureRemoteSpeechOutput() != nil {
				t.Fatal("controller acquired observation")
			}
			session.RevokeRemote()
			invokeSessionSDK(t, session, "event", nil) // Revocation.
			lease = speech_fixture_lease(t, session, "controlled", []string{"output-observe"})
			output := manager.CaptureRemoteSpeechOutput()
			if output == nil {
				t.Fatal("live controlled output unavailable")
			}
			text := "  " + strings.Repeat("é\x00世界\t", 170) + "  "
			if err := output.PublishText(text, true); err != nil {
				t.Fatal(err)
			}
			var received strings.Builder
			frames, ends := 0, 0
			for ends == 0 {
				var event tentaclewire.SessionEvent
				if err := json.Unmarshal(invokeSessionSDK(t, session, "event", nil), &event); err != nil {
					t.Fatal(err)
				}
				frames++
				if event.Kind != "remote.output" || event.Lease != lease.Lease || event.Epoch != lease.Epoch ||
					event.Speech == nil || !tentaclewire.ValidRemoteSpeech(*event.Speech) {
					t.Fatal("invalid observed event", event)
				}
				for _, part := range event.Speech.Parts {
					if part.End {
						ends++
					} else {
						if !part.Characters || part.Language != "" {
							t.Fatal("changed character/default-language state")
						}
						received.WriteString(part.Text)
					}
				}
			}
			if frames < 2 || ends != 1 || received.String() != text {
				t.Fatal("changed split text or boundaries")
			}
			if err := output.Cancel(); err != nil {
				t.Fatal(err)
			}
			var event tentaclewire.SessionEvent
			if json.Unmarshal(invokeSessionSDK(t, session, "event", nil), &event) != nil || event.Speech.Kind != "cancel" {
				t.Fatal("lost cancellation order")
			}
			session.RevokeRemote()
			invokeSessionSDK(t, session, "event", nil)
			_ = speech_fixture_lease(t, session, "controlled", []string{"output-observe"})
			if !errors.Is(output.PublishText("old", false), ErrPermission) || !errors.Is(output.Cancel(), ErrPermission) {
				t.Fatal("old ticket followed replacement lease")
			}
		})
	}
}

// No pipe writer: prove that a split output reserves all queue space before
// publishing a prefix, including encoded-byte and in-flight accounting.
func TestRemoteSpeechOutputAdmissionAndPolicy(t *testing.T) {
	for _, mode := range []string{"slots", "bytes", "frames", "invalid", "policy", "expired", "pending"} {
		t.Run(mode, func(t *testing.T) {
			manager := test_manager(t, DefaultLimits(), nil)
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			outputContext, revoke := context.WithCancel(ctx)
			defer revoke()
			session := &Session{manager: manager, spec: Spec{SessionEvents: true}, ctx: ctx, cancel: cancel,
				outputContext: outputContext, revokeOutput: revoke, state: Active,
				outbound: make(chan outgoing, 2), revoked: make(chan struct{})}
			accessContext, retire := context.WithCancel(ctx)
			defer retire()
			session.remote = &remoteAccess{lease: strings.Repeat("a", 32), epoch: 1, role: "controlled",
				grants: []string{"output-observe"}, ctx: accessContext, cancel: retire, deadline: time.Now().Add(time.Minute)}
			manager.remoteOwner = session
			output := manager.CaptureRemoteSpeechOutput()
			text := strings.Repeat("x", 800)
			want := ErrLimit
			switch mode {
			case "slots":
				session.outbound <- outgoing{frame: []byte("existing")}
			case "bytes":
				session.queuedBytes = tentaclewire.SessionQueueBytes - 1
			case "frames":
				text = strings.Repeat("x", 768*tentaclewire.RemoteSpeechQueueSize+1)
			case "invalid":
				text = "\xff"
			case "policy":
				manager.remotePolicy = NewRemotePolicy()
				session.remote.ctx, session.remote.cancel = context.WithCancel(manager.remotePolicy.Context())
				output.ctx = session.remote.ctx
				defer session.remote.cancel()
				manager.remotePolicy.SetAvailable(false)
				manager.remotePolicy.SetAvailable(true)
				want = ErrPermission
			case "expired":
				session.remote.deadline = time.Now().Add(-time.Second)
				want = ErrPermission
			case "pending":
				session.remote.lease = ""
				want = ErrPermission
			}
			before := len(session.outbound)
			if err := output.PublishText(text, false); !errors.Is(err, want) {
				t.Fatalf("publish = %v, want %v", err, want)
			}
			if len(session.outbound) != before || session.eventSequence != 0 {
				t.Fatal("failed admission published a prefix or consumed event sequence")
			}
			if want == ErrLimit && (session.state != Failed || accessContext.Err() == nil) {
				t.Fatal("overflow retained live stream")
			}
			if want == ErrPermission && manager.CaptureRemoteSpeechOutput() != nil {
				t.Fatal("captured invalid output authority")
			}
		})
	}
}