storm/cthulhu
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Assemble Remote speech before reader delivery

Browse Source
This commit is contained in:
ak47 committed 2026-11-04 14:33:54 -05:00
1 parent f23e89ed93
commit 205cb27483
9 files changed
+492 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/tentacle/remote.go
+16 -11
View File
@@ -27,17 +27,20 @@ func (session *Session) RemoteGrants() []string { return slices.Clone(session.sp
// Owned by Session.mutex. No network identity or secret belongs here.
type remoteAccess struct {
action string
lease string
epoch uint64
sequence uint64
role string
grants []string
ctx context.Context
cancel context.CancelFunc
deadline time.Time
timer *time.Timer
stopPolicy func() bool
action string
lease string
epoch uint64
sequence uint64
role string
grants []string
ctx context.Context
cancel context.CancelFunc
deadline time.Time
timer *time.Timer
stopPolicy func() bool
speechBatch *remoteSpeechBatch
speechItems int
speechBytes int
}
// AuthorizeRemote is the unrestricted protocol-fixture hook. Reader managers
@@ -160,6 +163,8 @@ func (session *Session) revoke_remote_locked(reason string, notify bool) {
}
session.remote = nil
access.cancel()
access.speechBatch = nil
access.speechItems, access.speechBytes = 0, 0
if session.remoteSpeech != nil {
drain:
for {
internal/tentacle/remote_speech.go
+52 -4
View File
@@ -7,6 +7,7 @@ import (
"context"
"encoding/json"
"slices"
"sync"
"time"
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
@@ -19,7 +20,18 @@ import (
type RemoteSpeechDelivery struct {
Context context.Context
Lease tentaclewire.RemoteLease
Speech tentaclewire.RemoteSpeech
// Speeches is one complete effect/utterance, including all IPC fragments.
// The owner applies priority once and must not deliver a partial assembly.
Speeches []tentaclewire.RemoteSpeech
// Release returns the retained frame/byte budget after delivery or discard.
// It is idempotent and safe after revocation. Dequeue alone frees no budget.
Release func()
}
type remoteSpeechBatch struct {
id uint64
speeches []tentaclewire.RemoteSpeech
bytes int
}
// EnableRemoteSpeech installs one bounded mailbox, explicitly from reader code.
@@ -67,7 +79,7 @@ func (session *Session) PublishRemoteSpeech(lease tentaclewire.RemoteLease, orig
func (session *Session) remote_present_locked(params json.RawMessage) (any, *RPCError) {
var request tentaclewire.RemotePresentation
if decode_strict(params, &request) != nil || !tentaclewire.ValidRemoteSpeech(request.Speech) || len(request.Lease) != 32 || request.Epoch == 0 || request.Sequence == 0 {
if decode_strict(params, &request) != nil {
return nil, &RPCError{Code: -32602, Message: "Invalid session parameters"}
}
lease := tentaclewire.RemoteLease{Lease: request.Lease, Epoch: request.Epoch}
@@ -78,9 +90,45 @@ func (session *Session) remote_present_locked(params json.RawMessage) (any, *RPC
if session.remoteSpeech == nil {
return nil, &RPCError{Code: -32601, Message: "Remote speech owner unavailable"}
}
batch := access.speechBatch
if batch == nil {
if request.Utterance != 0 && request.Utterance != request.Sequence {
return nil, &RPCError{Code: -32602, Message: "Invalid utterance assembly"}
}
batch = &remoteSpeechBatch{id: request.Utterance}
} else if request.Utterance != batch.id || request.Speech.Priority != batch.speeches[0].Priority {
// No interleaved speech, cancel, pause, or changed priority may expose
// an incomplete utterance. Heartbeats may still advance the sequence.
return nil, &RPCError{Code: -32602, Message: "Invalid utterance assembly"}
}
data, _ := json.Marshal(request.Speech)
if access.speechItems == tentaclewire.RemoteSpeechQueueSize || access.speechBytes+len(data) > tentaclewire.RemoteSpeechQueueBytes {
session.revoke_remote_locked("local", true)
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
}
access.speechItems++
access.speechBytes += len(data)
batch.speeches = append(batch.speeches, request.Speech)
batch.bytes += len(data)
access.sequence = request.Sequence
if request.Utterance != 0 && !request.Final {
access.speechBatch = batch
return struct{}{}, nil
}
access.speechBatch = nil
var released sync.Once
delivery := RemoteSpeechDelivery{Context: access.ctx, Lease: lease, Speeches: batch.speeches, Release: func() {
released.Do(func() {
session.mutex.Lock()
defer session.mutex.Unlock()
if session.remote == access {
access.speechItems -= len(batch.speeches)
access.speechBytes -= batch.bytes
}
})
}}
select {
case session.remoteSpeech <- RemoteSpeechDelivery{Context: access.ctx, Lease: lease, Speech: request.Speech}:
access.sequence = request.Sequence
case session.remoteSpeech <- delivery:
return struct{}{}, nil
default:
// A dropped cancel would let stale speech outlive the intended stream.
internal/tentacle/remote_speech_test.go
+211 -3
View File
@@ -5,14 +5,27 @@ 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()
action, err := session.AuthorizeRemote(role, grants)
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)
}
@@ -43,7 +56,7 @@ func TestRemoteSpeechSDKDeliveryAndRevocation(t *testing.T) {
t.Fatal(code)
}
first := <-deliveries
if first.Speech.Parts[0] != request.Speech.Parts[0] || first.Context.Err() != nil {
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 {
@@ -60,7 +73,7 @@ func TestRemoteSpeechSDKDeliveryAndRevocation(t *testing.T) {
if _, code := remoteCall(t, session, "remote.present", request); code != 0 {
t.Fatal(code)
}
if (<-deliveries).Speech.Kind != "cancel" || (<-deliveries).Speech.Kind != "pause" {
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.
@@ -149,3 +162,198 @@ func TestRemoteSpeechQueueOverflowRetiresLease(t *testing.T) {
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)
}
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 !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")
}
})
}
}