ak47
49be8309fe
Attach incoming software speech to reader sessions with lease checks, local output priority and isolated cancellation. Preserve every document fragment source for privacy checks and export reviewed browse current-line reading, spelling and phonetic output. Verified focused reading and Remote race tests, complete review and external Remote race suites, private relay integration, speech regressions, affected-package vet, reader build and Tentacles make check. Audible speech, physical input/Braille and live interoperability remain unverified.
364 lines
14 KiB
Go
364 lines
14 KiB
Go
// 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")
|
|
}
|
|
})
|
|
}
|
|
}
|