Tentacle enhancements.

This commit is contained in:
Storm Dragon committed 2026-11-04 13:44:13 -05:00
1 parent 4f9b89e38b
commit 9e15f8652d
12 files changed
+619 -22

No files matched your search

+14 -2
View File
@@ -134,8 +134,8 @@ func (session *Session) RevokeRemoteEpoch(epoch uint64) {
func (session *Session) RemoteContext(lease tentaclewire.RemoteLease, grant string) (context.Context, error) {
session.mutex.Lock()
defer session.mutex.Unlock()
access := session.remote
if session.state != Active || access == nil || access.lease == "" || access.lease != lease.Lease || access.epoch != lease.Epoch || !time.Now().Before(access.deadline) || access.ctx.Err() != nil || !slices.Contains(access.grants, grant) {
access := session.remote_access_locked(lease, grant)
if access == nil {
return nil, ErrPermission
}
return access.ctx, nil
@@ -160,6 +160,16 @@ func (session *Session) revoke_remote_locked(reason string, notify bool) {
}
session.remote = nil
access.cancel()
if session.remoteSpeech != nil {
drain:
for {
select {
case <-session.remoteSpeech:
default:
break drain
}
}
}
if access.stopPolicy != nil {
access.stopPolicy()
}
@@ -235,6 +245,8 @@ func (session *Session) remote_call_locked(method string, params json.RawMessage
denied := &RPCError{Code: -32001, Message: "Session permission denied"}
invalid := &RPCError{Code: -32602, Message: "Invalid session parameters"}
switch method {
case "remote.present":
return session.remote_present_locked(params)
case "remote.open":
var request tentaclewire.RemoteOpen
if decode_strict(params, &request) != nil || len(request.Action) != 32 || request.Epoch == 0 || !tentaclewire.ValidRoleGrants(request.Role, request.Grants) {
+90
View File
@@ -0,0 +1,90 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package tentacle
import (
"context"
"encoding/json"
"slices"
"time"
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
)
// RemoteSpeechDelivery is reader-owned work. The owner must check Context and
// RemoteContext again at delivery, apply focus/security/output policy, and tag
// the result as remote so neither observation nor remote cancel affects local
// status. Admission is not evidence of backend delivery or audible speech.
type RemoteSpeechDelivery struct {
Context context.Context
Lease tentaclewire.RemoteLease
Speech tentaclewire.RemoteSpeech
}
// EnableRemoteSpeech installs one bounded mailbox, explicitly from reader code.
// Without an owner, remote.present is unavailable. No production speech worker
// is bound yet. The owner selects Session.Done as well as this channel.
func (session *Session) EnableRemoteSpeech() <-chan RemoteSpeechDelivery {
session.mutex.Lock()
defer session.mutex.Unlock()
if session.remoteSpeech == nil {
session.remoteSpeech = make(chan RemoteSpeechDelivery, tentaclewire.RemoteSpeechQueueSize)
}
return session.remoteSpeech
}
func (session *Session) remote_access_locked(lease tentaclewire.RemoteLease, grant string) *remoteAccess {
access := session.remote
if session.state != Active || access == nil || access.lease == "" || access.lease != lease.Lease || access.epoch != lease.Epoch || !time.Now().Before(access.deadline) || access.ctx.Err() != nil || !slices.Contains(access.grants, grant) {
return nil
}
return access
}
// PublishRemoteSpeech accepts only policy-resolved local output from a reader
// owner. The origin is supplied by core, never by child RPC or network input.
// Remote output, local status, and sensitive output must use other origins.
// The owner must exclude password/secret output before calling this method.
func (session *Session) PublishRemoteSpeech(lease tentaclewire.RemoteLease, origin string, speech tentaclewire.RemoteSpeech) error {
if origin != "local" || !tentaclewire.ValidRemoteSpeech(speech) {
return ErrPermission
}
session.mutex.Lock()
defer session.mutex.Unlock()
access := session.remote_access_locked(lease, "output-observe")
if access == nil || access.role != "controlled" {
return ErrPermission
}
event := tentaclewire.SessionEvent{Kind: "remote.output", Lease: lease.Lease, Epoch: lease.Epoch, Speech: &speech}
if !session.remote_event_locked(&event) {
// Shared lifecycle IPC cannot lose revocation or cancel order.
session.fail_locked(ErrLimit)
return ErrLimit
}
return nil
}
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 {
return nil, &RPCError{Code: -32602, Message: "Invalid session parameters"}
}
lease := tentaclewire.RemoteLease{Lease: request.Lease, Epoch: request.Epoch}
access := session.remote_access_locked(lease, "remote-present")
if access == nil || access.role != "controller" || request.Sequence <= access.sequence {
return nil, &RPCError{Code: -32001, Message: "Session permission denied"}
}
if session.remoteSpeech == nil {
return nil, &RPCError{Code: -32601, Message: "Remote speech owner unavailable"}
}
select {
case session.remoteSpeech <- RemoteSpeechDelivery{Context: access.ctx, Lease: lease, Speech: request.Speech}:
access.sequence = request.Sequence
return struct{}{}, nil
default:
// A dropped cancel would let stale speech outlive the intended stream.
session.revoke_remote_locked("local", true)
return nil, &RPCError{Code: -32002, Message: "Remote speech queue full"}
}
}
+151
View File
@@ -0,0 +1,151 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package tentacle
import (
"encoding/json"
"testing"
"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)
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.Speech.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).Speech.Kind != "cancel" || (<-deliveries).Speech.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:
}
}
+6
View File
@@ -433,6 +433,12 @@ func TestSessionSDKRejectsMalformedHostMessages(t *testing.T) {
overflow += strings.Replace(event, `"sequence":1`, `"sequence":`+strconv.Itoa(sequence), 1)
}
cases["event overflow"] = overflow
outputEvent := `{"jsonrpc":"2.0","method":"session.event","params":{"sequence":1,"kind":"remote.output","lease":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","epoch":1,"speech":{"kind":"cancel"}}}` + "\n"
cases["output without lease"] = initialize + activate + strings.Replace(outputEvent, `"lease":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",`, "", 1)
cases["output with role"] = initialize + activate + strings.Replace(outputEvent, `"epoch":1`, `"epoch":1,"role":"controller"`, 1)
cases["null speech"] = initialize + activate + strings.Replace(outputEvent, `{"kind":"cancel"}`, `null`, 1)
cases["malformed speech"] = initialize + activate + strings.Replace(outputEvent, `{"kind":"cancel"}`, `{"kind":"pause","paused":null}`, 1)
cases["speech on action"] = initialize + activate + strings.Replace(event, `"epoch":1`, `"epoch":1,"speech":{"kind":"cancel"}`, 1)
for _, language := range []string{"go", "python"} {
for name, input := range cases {
t.Run(language+"/"+name, func(t *testing.T) {
+1
View File
@@ -76,6 +76,7 @@ type Session struct {
remoteEpoch uint64
remote *remoteAccess
remoteApproval *RemoteApproval
remoteSpeech chan RemoteSpeechDelivery
}
func (session *Session) Ready() <-chan struct{} { return session.ready }
+106
View File
@@ -0,0 +1,106 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package tentaclewire
import (
"bytes"
"encoding/json"
"unicode/utf8"
)
// Leave room for lease metadata inside the existing 8192-byte call bound.
const MaxRemoteSpeechBytes = 6144
const RemoteSpeechQueueSize = 64
// RemoteSpeechPart carries resolved character/language state for each fragment.
// End is an utterance boundary, not text. Whitespace is significant.
type RemoteSpeechPart struct {
Text string `json:"text,omitempty"`
Characters bool `json:"characters,omitempty"`
Language string `json:"language,omitempty"`
End bool `json:"end,omitempty"`
}
type RemoteSpeech struct {
Kind string `json:"kind"` // speak, cancel, pause
Priority int `json:"priority,omitempty"` // normal=0, next=1, now=2
Parts []RemoteSpeechPart `json:"parts,omitempty"`
Paused *bool `json:"paused,omitempty"`
}
type RemotePresentation struct {
RemoteOperation
Speech RemoteSpeech `json:"speech"`
}
func reject_speech_nulls(data []byte) error {
var fields map[string]json.RawMessage
if json.Unmarshal(data, &fields) != nil || fields == nil {
return ErrProtocol
}
for _, value := range fields {
if bytes.Equal(bytes.TrimSpace(value), []byte("null")) {
return ErrProtocol
}
}
return nil
}
func (part *RemoteSpeechPart) UnmarshalJSON(data []byte) error {
type plain RemoteSpeechPart
var value plain
if reject_speech_nulls(data) != nil || DecodeStrict(data, &value) != nil {
return ErrProtocol
}
*part = RemoteSpeechPart(value)
return nil
}
func (speech *RemoteSpeech) UnmarshalJSON(data []byte) error {
type plain RemoteSpeech
var value plain
if reject_speech_nulls(data) != nil || DecodeStrict(data, &value) != nil || !ValidRemoteSpeech(RemoteSpeech(value)) {
return ErrProtocol
}
*speech = RemoteSpeech(value)
return nil
}
// ValidRemoteSpeech also limits encoded bytes: escaped text must fit in IPC.
func ValidRemoteSpeech(speech RemoteSpeech) bool {
switch speech.Kind {
case "speak":
if speech.Priority < 0 || speech.Priority > 2 || speech.Paused != nil || len(speech.Parts) == 0 || len(speech.Parts) > 128 {
return false
}
for _, part := range speech.Parts {
if part.End {
if part.Text != "" || part.Characters || part.Language != "" {
return false
}
continue
}
if part.Text == "" || !utf8.ValidString(part.Text) || len(part.Language) > 64 {
return false
}
for _, char := range part.Language {
if !(char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z' || char >= '0' && char <= '9' || char == '-' || char == '_') {
return false
}
}
}
case "cancel":
if speech.Priority != 0 || len(speech.Parts) != 0 || speech.Paused != nil {
return false
}
case "pause":
if speech.Priority != 0 || len(speech.Parts) != 0 || speech.Paused == nil {
return false
}
default:
return false
}
data, err := json.Marshal(speech)
return err == nil && len(data) <= MaxRemoteSpeechBytes
}
@@ -0,0 +1,92 @@
// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later
package tentaclewire
import (
"encoding/json"
"os/exec"
"path/filepath"
"strings"
"testing"
)
func TestRemoteSpeechSDKValidation(t *testing.T) {
cases := []struct {
data string
valid bool
}{
{`{"kind":"speak","parts":[{"text":" hello\t世界 ","characters":true,"language":"en_US"},{"end":true}]}`, true},
{`{"kind":"cancel"}`, true},
{`{"kind":"pause","paused":false}`, true},
{`{"kind":"pause","paused":true}`, true},
{`{"kind":"speak","priority":2,"parts":[{"text":"x"}]}`, true},
{`{"kind":"speak","priority":3,"parts":[{"text":"x"}]}`, false},
{`{"kind":"speak","priority":true,"parts":[{"text":"x"}]}`, false},
{`{"kind":"speak","parts":[]}`, false},
{`{"kind":"speak","parts":[{}]}`, false},
{`{"kind":"speak","parts":[{"text":null}]}`, false},
{`{"kind":"speak","parts":[{"text":"x","end":true}]}`, false},
{`{"kind":"speak","parts":[{"text":"x","language":"../en"}]}`, false},
{`{"kind":"speak","parts":[{"text":"x","voice":"peer path"}]}`, false},
{`{"kind":"speak","parts":[{"text":"x"}],"paused":false}`, false},
{`{"kind":"cancel","parts":null}`, false},
{`{"kind":"cancel","parts":[{"text":"x"}]}`, false},
{`{"kind":"cancel","paused":false}`, false},
{`{"kind":"pause"}`, false},
{`{"kind":"pause","paused":null}`, false},
{`{"kind":"pause","paused":0}`, false},
{`{"kind":"wave","path":"anything"}`, false},
{`{"kind":"cancel","priority":null}`, false},
{`{"kind":"cancel","extra":1}`, false},
{`{"kind":"speak","parts":[{"text":"` + strings.Repeat("x", MaxRemoteSpeechBytes) + `"}]}`, false},
{`{"kind":"speak","parts":[{"text":"` + strings.Repeat("<", 1100) + `"}]}`, false},
{`{"kind":"speak","parts":[{"text":"` + strings.Repeat("\u2028", 1100) + `"}]}`, false},
}
var pythonCases []any
for _, item := range cases {
var speech RemoteSpeech
valid := DecodeStrict([]byte(item.data), &speech) == nil && ValidRemoteSpeech(speech)
if valid != item.valid {
t.Fatalf("Go validation: %.100s: got %v", item.data, valid)
}
var value any
if err := json.Unmarshal([]byte(item.data), &value); err != nil {
t.Fatal(err)
}
pythonCases = append(pythonCases, map[string]any{"speech": value, "valid": item.valid})
}
path, err := filepath.Abs("../../sdk/python")
if err != nil {
t.Fatal(err)
}
script := `import json, sys
sys.path.insert(0, sys.argv[1])
from cthulhu_tentacle.session import valid_remote_speech
for case in json.load(sys.stdin):
assert valid_remote_speech(case['speech']) == case['valid'], str(case)[:150]
`
data, _ := json.Marshal(pythonCases)
command := exec.Command("python3", "-B", "-c", script, path)
command.Stdin = strings.NewReader(string(data))
if output, err := command.CombinedOutput(); err != nil {
t.Fatalf("Python validation: %v: %s", err, output)
}
}
func TestRemoteSpeechEncodedBoundary(t *testing.T) {
value := RemoteSpeech{Kind: "speak", Parts: []RemoteSpeechPart{{Text: "x"}}}
data, _ := json.Marshal(value)
value.Parts[0].Text = strings.Repeat("x", MaxRemoteSpeechBytes-len(data)+1)
if !ValidRemoteSpeech(value) {
t.Fatal("rejected exact bound")
}
value.Parts[0].Text += "x"
if ValidRemoteSpeech(value) {
t.Fatal("accepted above bound")
}
value.Parts[0].Text = string([]byte{0xff})
if ValidRemoteSpeech(value) {
t.Fatal("accepted invalid UTF-8")
}
}
+15 -12
View File
@@ -63,17 +63,18 @@ func ValidNegotiation(minor int, apis, capabilities []string) bool {
return a["remote-access"] == hasRemote
}
// SessionEvent is deliberately limited to lifecycle metadata. Output/input
// payloads will require their own typed validators before they can be enabled.
// SessionEvent carries lifecycle metadata and explicitly typed output. Output
// observation still requires a reader owner and a live role-bound lease.
type SessionEvent struct {
Sequence uint64 `json:"sequence"`
Kind string `json:"kind"`
Action string `json:"action,omitempty"`
Lease string `json:"lease,omitempty"`
Epoch uint64 `json:"epoch"`
Role string `json:"role,omitempty"`
Grants []string `json:"grants,omitempty"`
Reason string `json:"reason,omitempty"`
Sequence uint64 `json:"sequence"`
Kind string `json:"kind"`
Action string `json:"action,omitempty"`
Lease string `json:"lease,omitempty"`
Epoch uint64 `json:"epoch"`
Role string `json:"role,omitempty"`
Grants []string `json:"grants,omitempty"`
Reason string `json:"reason,omitempty"`
Speech *RemoteSpeech `json:"speech,omitempty"`
}
func ValidRole(role string) bool { return role == "controller" || role == "controlled" }
@@ -96,9 +97,11 @@ func ValidSessionEvent(event SessionEvent) bool {
}
switch event.Kind {
case "remote.action":
return len(event.Action) == 32 && event.Lease == "" && event.Reason == "" && ValidRoleGrants(event.Role, event.Grants)
return event.Speech == nil && len(event.Action) == 32 && event.Lease == "" && event.Reason == "" && ValidRoleGrants(event.Role, event.Grants)
case "remote.revoked":
return event.Action == "" && len(event.Lease) == 32 && event.Role == "" && len(event.Grants) == 0 && slices.Contains([]string{"closed", "expired", "local", "replaced"}, event.Reason)
return event.Speech == nil && event.Action == "" && len(event.Lease) == 32 && event.Role == "" && len(event.Grants) == 0 && slices.Contains([]string{"closed", "expired", "local", "replaced"}, event.Reason)
case "remote.output":
return event.Action == "" && len(event.Lease) == 32 && event.Role == "" && len(event.Grants) == 0 && event.Reason == "" && event.Speech != nil && ValidRemoteSpeech(*event.Speech)
}
return false
}