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

package tentacle

import (
	"encoding/json"
	"reflect"
	"strings"
	"testing"
	"time"

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

func speech_fixture_lease(t *testing.T, session *Session, role string, grants []string) tentaclewire.RemoteLease {
	t.Helper()
	var action tentaclewire.SessionEvent
	var err error
	if session.manager.remotePolicy == nil {
		action, err = session.AuthorizeRemote(role, grants)
	} else {
		approval, failure := session.BeginRemoteApproval(role, grants)
		if failure != nil {
			t.Fatal(failure)
		}
		action, err = session.AuthorizeRemoteWithApproval(approval, role, grants)
	}
	if err != nil {
		t.Fatal(err)
	}
	invokeSessionSDK(t, session, "event", nil)
	raw, code := remoteCall(t, session, "remote.open", tentaclewire.RemoteOpen{Action: action.Action, Epoch: action.Epoch, Role: role, Grants: grants})
	if code != 0 {
		t.Fatal("open", code)
	}
	var lease tentaclewire.RemoteLease
	if err := json.Unmarshal(raw, &lease); err != nil {
		t.Fatal(err)
	}
	return lease
}

func TestRemoteSpeechSDKDeliveryAndRevocation(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")
			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			request := tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch, Sequence: 1}, Speech: tentaclewire.RemoteSpeech{Kind: "speak", Parts: []tentaclewire.RemoteSpeechPart{{Text: "  text\t世界  ", Characters: true, Language: "en_US"}}}}
			if _, code := remoteCall(t, session, "remote.present", request); code != -32601 {
				t.Fatal("accepted without reader owner", code)
			}
			deliveries := session.EnableRemoteSpeech()
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			first := <-deliveries
			if first.Speeches[0].Parts[0] != request.Speech.Parts[0] || first.Context.Err() != nil {
				t.Fatal("changed speech or invalid lifetime")
			}
			if _, code := remoteCall(t, session, "remote.present", request); code != -32001 {
				t.Fatal("accepted replay", code)
			}
			request.Sequence++
			request.Speech = tentaclewire.RemoteSpeech{Kind: "cancel"}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			request.Sequence++
			paused := true
			request.Speech = tentaclewire.RemoteSpeech{Kind: "pause", Paused: &paused}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			if (<-deliveries).Speeches[0].Kind != "cancel" || (<-deliveries).Speeches[0].Kind != "pause" {
				t.Fatal("lost cancel ordering")
			}
			// Heartbeat shares the operation sequence, but speech is not a heartbeat.
			if _, code := remoteCall(t, session, "remote.heartbeat", request.RemoteOperation); code != -32001 {
				t.Fatal("heartbeat replay", code)
			}
			request.Sequence++
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			session.RevokeRemote()
			if first.Context.Err() == nil {
				t.Fatal("accepted work survived revoke")
			}
			if _, err := session.RemoteContext(lease, "remote-present"); err == nil {
				t.Fatal("delivery-time authorization survived revoke")
			}
			select {
			case <-deliveries:
				t.Fatal("retained revoked work")
			default:
			}
			request.Sequence++
			if _, code := remoteCall(t, session, "remote.present", request); code != -32001 {
				t.Fatal("accepted stale lease", code)
			}
		})
	}
}

func TestRemoteSpeechSDKOutputAndOrigins(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")
			lease := speech_fixture_lease(t, session, "controlled", []string{"output-observe"})
			value := tentaclewire.RemoteSpeech{Kind: "speak", Priority: 2, Parts: []tentaclewire.RemoteSpeechPart{{Text: " whitespace \n "}}}
			for _, origin := range []string{"remote", "status", "secret", ""} {
				if err := session.PublishRemoteSpeech(lease, origin, value); err == nil {
					t.Fatal("exported origin", origin)
				}
			}
			if err := session.PublishRemoteSpeech(lease, "local", value); err != nil {
				t.Fatal(err)
			}
			var event tentaclewire.SessionEvent
			if err := json.Unmarshal(invokeSessionSDK(t, session, "event", nil), &event); err != nil {
				t.Fatal(err)
			}
			if event.Kind != "remote.output" || event.Lease != lease.Lease || event.Epoch != lease.Epoch || event.Speech == nil || event.Speech.Parts[0] != value.Parts[0] {
				t.Fatal("changed output event", event)
			}
			if _, code := remoteCall(t, session, "remote.present", tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch, Sequence: 1}, Speech: value}); code != -32001 {
				t.Fatal("wrong-direction grant", code)
			}
			session.RevokeRemote()
			if err := session.PublishRemoteSpeech(lease, "local", value); err == nil {
				t.Fatal("exported after revoke")
			}
		})
	}
}

func TestRemoteSpeechQueueOverflowRetiresLease(t *testing.T) {
	manager := test_manager(t, DefaultLimits(), nil)
	session := startSessionSDK(t, manager, "go", "speech")
	lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
	deliveries := session.EnableRemoteSpeech()
	request := tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch}, Speech: tentaclewire.RemoteSpeech{Kind: "cancel"}}
	for index := 1; index <= tentaclewire.RemoteSpeechQueueSize; index++ {
		request.Sequence = uint64(index)
		if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
			t.Fatal(index, code)
		}
	}
	request.Sequence++
	if _, code := remoteCall(t, session, "remote.present", request); code != -32002 {
		t.Fatal("queue did not fail closed", code)
	}
	if _, err := session.RemoteContext(lease, "remote-present"); err == nil {
		t.Fatal("overflow retained authorization")
	}
	select {
	case <-deliveries:
		t.Fatal("overflow retained pending speech")
	default:
	}
}

func TestRemoteSpeechSDKAtomicUtterance(t *testing.T) {
	for _, language := range []string{"go", "python"} {
		t.Run(language, func(t *testing.T) {
			session := startSessionSDK(t, test_manager(t, DefaultLimits(), nil), language, "speech")
			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			deliveries := session.EnableRemoteSpeech()
			frames := []tentaclewire.RemoteSpeech{
				{Kind: "speak", Priority: 2, Parts: []tentaclewire.RemoteSpeechPart{{Text: "  prefix\t", Characters: true, Language: "fr"}}},
				{Kind: "speak", Priority: 2, Parts: []tentaclewire.RemoteSpeechPart{{Text: "世界  "}}},
				{Kind: "speak", Priority: 2, Parts: []tentaclewire.RemoteSpeechPart{{End: true}}},
			}
			request := tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch, Sequence: 1}, Utterance: 1, Speech: frames[0]}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			firstAdmitted := time.Now()
			request.Sequence++
			if _, code := remoteCall(t, session, "remote.heartbeat", request.RemoteOperation); code != 0 {
				t.Fatal("heartbeat could not run during assembly", code)
			}
			request.Sequence++
			request.Speech = frames[1]
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			select {
			case <-deliveries:
				t.Fatal("exposed incomplete utterance")
			default:
			}
			if _, code := remoteCall(t, session, "remote.present", request); code != -32001 {
				t.Fatal("accepted fragment replay", code)
			}
			request.Sequence++
			for _, invalid := range []tentaclewire.RemotePresentation{
				{RemoteOperation: request.RemoteOperation, Speech: tentaclewire.RemoteSpeech{Kind: "cancel"}},
				{RemoteOperation: request.RemoteOperation, Speech: frames[2], Utterance: request.Sequence, Final: true},
				{RemoteOperation: request.RemoteOperation, Speech: tentaclewire.RemoteSpeech{Kind: "speak", Parts: []tentaclewire.RemoteSpeechPart{{Text: "priority changed"}}}, Utterance: 1, Final: true},
			} {
				if _, code := remoteCall(t, session, "remote.present", invalid); code != -32602 {
					t.Fatal("accepted interleaving or changed priority", code)
				}
			}
			request.Speech, request.Final = frames[2], true
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			delivery := <-deliveries
			if delivery.Received.IsZero() || delivery.Received.After(firstAdmitted) {
				t.Fatal("finalization refreshed the original assembly admission time")
			}
			if !reflect.DeepEqual(delivery.Speeches, frames) || delivery.Lease != lease || delivery.Context.Err() != nil {
				t.Fatal("assembly changed semantics or ownership")
			}
			request.Sequence++
			request.Utterance, request.Final, request.Speech = 0, false, tentaclewire.RemoteSpeech{Kind: "cancel"}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			cancel := <-deliveries
			if len(cancel.Speeches) != 1 || cancel.Speeches[0].Kind != "cancel" {
				t.Fatal("cancel lost ordering")
			}
			session.mutex.Lock()
			if session.remote.speechItems != 4 || session.remote.speechBatch != nil {
				t.Error("dequeue released in-flight budget")
			}
			session.mutex.Unlock()
			delivery.Release()
			delivery.Release()
			cancel.Release()
			session.mutex.Lock()
			defer session.mutex.Unlock()
			if session.remote.speechItems != 0 || session.remote.speechBytes != 0 {
				t.Fatal("release leaked budget or counted twice")
			}
		})
	}
}

func TestRemoteSpeechSDKAssemblyRetirement(t *testing.T) {
	for _, mode := range []string{"revoke", "replace", "close", "expiry", "policy"} {
		t.Run(mode, func(t *testing.T) {
			manager := test_manager(t, DefaultLimits(), nil)
			policy := NewRemotePolicy()
			if mode == "policy" {
				if err := manager.UseRemotePolicy(policy); err != nil {
					t.Fatal(err)
				}
			}
			session := startSessionSDK(t, manager, "go", "speech")
			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			deliveries := session.EnableRemoteSpeech()
			request := tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch, Sequence: 1}, Speech: tentaclewire.RemoteSpeech{Kind: "cancel"}}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			inFlight := <-deliveries
			request.Sequence, request.Utterance = 2, 2
			request.Speech = tentaclewire.RemoteSpeech{Kind: "speak", Parts: []tentaclewire.RemoteSpeechPart{{Text: "never a prefix"}}}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			session.mutex.Lock()
			old := session.remote
			session.mutex.Unlock()
			switch mode {
			case "revoke":
				session.RevokeRemote()
			case "replace":
				speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			case "close":
				request.Sequence++
				if _, code := remoteCall(t, session, "remote.close", request.RemoteOperation); code != 0 {
					t.Fatal(code)
				}
			case "expiry":
				session.mutex.Lock()
				session.refresh_remote_locked(time.Millisecond)
				session.mutex.Unlock()
				wait_signal(t, old.ctx.Done())
			case "policy":
				// Cancellation cannot wait for the session mutex; lock/handoff
				// invalidates work before asynchronous mailbox cleanup.
				session.mutex.Lock()
				policy.SetAvailable(false)
				cancelled := old.ctx.Err() != nil
				session.mutex.Unlock()
				if !cancelled {
					t.Fatal("policy cancellation waited for the session")
				}
				invokeSessionSDK(t, session, "event", nil) // Wait for retirement notification.
			}
			inFlight.Release() // A stale release must not debit a replacement lease.
			if inFlight.Context.Err() == nil {
				t.Fatal("in-flight delivery survived retirement")
			}
			request.Sequence++
			request.Final = true
			if _, code := remoteCall(t, session, "remote.present", request); code != -32001 {
				t.Fatal("stale fragment committed after retirement", code)
			}
			session.mutex.Lock()
			defer session.mutex.Unlock()
			if old.speechBatch != nil || old.speechItems != 0 || old.speechBytes != 0 || len(deliveries) != 0 {
				t.Fatal("retired assembly retained work")
			}
			if session.remote != nil && (session.remote.speechItems != 0 || session.remote.speechBytes != 0) {
				t.Fatal("stale release changed replacement budget")
			}
		})
	}
}

func TestRemoteSpeechSDKAssemblyBudget(t *testing.T) {
	for _, limit := range []string{"items", "bytes"} {
		t.Run(limit, func(t *testing.T) {
			session := startSessionSDK(t, test_manager(t, DefaultLimits(), nil), "python", "speech")
			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
			deliveries := session.EnableRemoteSpeech()
			text := "x"
			if limit == "bytes" {
				text = strings.Repeat("x", 5800)
			}
			request := tentaclewire.RemotePresentation{RemoteOperation: tentaclewire.RemoteOperation{Lease: lease.Lease, Epoch: lease.Epoch, Sequence: 1}, Speech: tentaclewire.RemoteSpeech{Kind: "speak", Parts: []tentaclewire.RemoteSpeechPart{{Text: text}}}}
			if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
				t.Fatal(code)
			}
			inFlight := <-deliveries // Retained until Release, even outside the mailbox.
			session.mutex.Lock()
			old := session.remote
			session.mutex.Unlock()
			request.Utterance = 2
			for index := 2; index <= tentaclewire.RemoteSpeechQueueSize+1; index++ {
				request.Sequence = uint64(index)
				_, code := remoteCall(t, session, "remote.present", request)
				if code == -32002 {
					if limit == "items" && index != tentaclewire.RemoteSpeechQueueSize+1 || limit == "bytes" && index >= tentaclewire.RemoteSpeechQueueSize {
						t.Fatal("wrong bound fired", index)
					}
					break
				}
				if code != 0 || index == tentaclewire.RemoteSpeechQueueSize+1 {
					t.Fatal("assembly overflow did not fail closed", index, code)
				}
			}
			if inFlight.Context.Err() == nil {
				t.Fatal("overflow left lease active")
			}
			inFlight.Release()
			session.mutex.Lock()
			defer session.mutex.Unlock()
			if session.remote != nil || old.speechBatch != nil || old.speechItems != 0 || old.speechBytes != 0 || len(deliveries) != 0 {
				t.Fatal("overflow exposed a prefix or retained work")
			}
		})
	}
}