diff --git a/docs/tentacle-remote-speech.md b/docs/tentacle-remote-speech.md index 111fa37..9209410 100644 --- a/docs/tentacle-remote-speech.md +++ b/docs/tentacle-remote-speech.md @@ -2,7 +2,10 @@ This increment adds typed speech IPC and an explicit reader-owned mailbox. It does not connect the speech worker, observe local output, or make Remote speak. -The external source command remains relay-only. Base reader revision: `4f9b89e` +The external coordinator now connects this IPC to relay speech; the production +reader still has no presentation owner or output observer. The current increment +adds atomic assembly of split incoming utterances before owner admission. +Original base reader revision: `4f9b89e` on `testing`; matching Tentacles base: `0c74038` on `master`, plus the pending connection-command changes. No active or installed reader was replaced. @@ -26,13 +29,35 @@ Child `remote.present` takes `RemotePresentation`: lease, epoch, monotonically increasing operation sequence and speech. The sequence is shared with heartbeat and close. Admission requires a live controller lease with `remote-present`. Without a reader owner calling `Session.EnableRemoteSpeech`, it returns `-32601`. -The queue holds at most 64 items, never blocks the IPC reader, and preserves FIFO -speech/cancel/pause order. Overflow revokes the entire lease and returns `-32002`; -it cannot drop a cancel while admitting newer speech. Speech does not refresh -the heartbeat deadline. +The queue never blocks the IPC reader and preserves FIFO speech/cancel/pause +order. Across incomplete assemblies, queued deliveries and work in flight, each +lease retains at most 64 speech frames and 256 KiB of canonical speech JSON. +Overflow revokes the entire lease and returns `-32002`; it cannot drop a cancel +while admitting newer speech. Speech does not refresh the heartbeat deadline. + +For split speech, `RemotePresentation` also accepts `utterance` and `final`. +Omitting both retains the single-call contract. A nonzero utterance identifies +one assembly and equals the first fragment's operation sequence. Later fragments +carry that same identifier and a strictly increasing operation sequence; the +last sets `final: true`. Only speak can be grouped, and every fragment must have +the same priority. Heartbeats may advance the shared sequence between fragments. +Only one assembly may be pending; interleaved effects, changed identifiers or +priorities return `-32602` without committing a prefix or consuming the sequence. +Replay still returns `-32001`. Null/unknown fields and invalid grouping metadata +are rejected; both SDKs expose matching presentation validators. + +No fragment reaches the owner before the final call succeeds. Finalization +queues one delivery containing every frame, in order. An unfinished assembly +stays bounded until completed or its lease ends; it has no independent renewal +or retry. Revocation, replacement, close, policy cancellation, process loss and +heartbeat expiry discard it. No relay protocol extension is involved. The owner receives `RemoteSpeechDelivery` containing a cancellation context, -lease and validated speech. It must select session shutdown, check the context +lease, `Speeches` slice and idempotent `Release` callback. The slice is one +complete effect/utterance; apply priority once, preserving all fragment states +and boundaries. The owner must call Release after delivery or discard and drop +its references. Dequeue alone does not return budget; stale releases cannot +debit a replacement lease. It must select session shutdown, check the context and `RemoteContext` at delivery, apply focus/secure-input/speech policy, keep remote cancellation separate from local status, and mark remote origins so they cannot be observed again. Revocation immediately cancels contexts and drains @@ -52,18 +77,24 @@ Tentacle must recheck lease/epoch and peer readiness before sending over a relay Run from this checkout: ```sh -GOCACHE=/tmp/cthulhu-cache GOPROXY=off go test -race ./internal/tentaclewire ./internal/tentacle ./sdk/... -count=1 +GOCACHE=/tmp/cthulhu-cache GOPROXY=off go test -race ./internal/tentaclewire ./internal/tentacle ./sdk/... -count=1 -timeout=240s ``` Real Go/Python child processes cover typed speech and observed events, absent owners, role/grant denial, replay, ordering, revocation and queue overflow. -Shared validation cases cover null/type/unknown-field rejection and escaped -byte bounds. Existing lifecycle/SDK tests exercise old 1.0 behavior, process -failure, timeout, event overflow and installation contracts. +Shared validation cases cover null/type/unknown-field rejection, grouping +metadata and escaped byte bounds. Assembly tests cover no-prefix admission, +interleaving, replay, priority consistency, heartbeats, in-flight accounting, +idempotent/stale release, byte/frame overflow and retirement. A private TLS +fixture in `TestRemoteConnectionCommandPrivateRelay` also passes long escaped +multilingual speech through the actual source command and reader host to one +complete test-owner delivery. Existing lifecycle/SDK tests exercise old 1.0 +behavior, process failure, timeout, event overflow and installation contracts. -Next: connect the policy-resolved speech worker and the external coordinator, -with origin ownership, secure/secret exclusion, delivery-time policy, local -priority, cancellation generations and atomic treatment of split utterances. +Next: connect the policy-resolved speech worker with origin ownership, +secure/secret exclusion, delivery-time policy, local +priority and cancellation generations. Atomic IPC assembly is implemented; +it does not establish backend priority/cancel behavior or audible delivery. Pause support or cancel-on-pause behavior and NVDA priority fidelity must be resolved there. The codec transports metadata without claiming backend behavior. Then verify normal reading, key echo, read-all, spelling, speech modes and local diff --git a/internal/app/tentacle_remote_connection_test.go b/internal/app/tentacle_remote_connection_test.go index 8a476b8..e3dd6ec 100644 --- a/internal/app/tentacle_remote_connection_test.go +++ b/internal/app/tentacle_remote_connection_test.go @@ -25,6 +25,7 @@ import ( "git.stormux.org/storm/cthulhu/internal/input" "git.stormux.org/storm/cthulhu/internal/interaction" "git.stormux.org/storm/cthulhu/internal/tentacle" + "git.stormux.org/storm/cthulhu/internal/tentaclewire" ) // This uses the source-built external executable and real host/SDK IPC. Only @@ -223,6 +224,7 @@ func TestRemoteConnectionCommandCancelsInspectionOnHandoff(t *testing.T) { func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { for _, role := range []string{"controller", "controlled"} { t.Run(role, func(t *testing.T) { + remoteText := " " + strings.Repeat("é\x00世界 ", 1500) + " " key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) if err != nil { t.Fatal(err) @@ -274,12 +276,39 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { t.Error("invalid relay join or wrong selected role") return } - io.WriteString(connection, `{"type":"channel_joined","channel":"synthetic-fixture","clients":[]}`+"\n") + peerRole := "master" + if role == "controller" { + peerRole = "slave" + } + io.WriteString(connection, `{"type":"channel_joined","channel":"synthetic-fixture","clients":[{"id":7,"connection_type":"`+peerRole+`"}]}`+"\n") + if role == "controller" { + json.NewEncoder(connection).Encode(map[string]any{"type": "speak", "origin": 7, "priority": 2, "sequence": []any{ + []any{"CharacterModeCommand", map[string]any{"state": true}}, + []any{"LangChangeCommand", map[string]any{"lang": "fr"}}, remoteText, + []any{"EndUtteranceCommand", map[string]any{}}, + }}) + } close(joined) io.Copy(io.Discard, reader) close(closed) }() runtime, forms, _ := remote_command_fixture(t) + // Only this fixture supplies a reader owner. Production remains + // disabled until output arbitration/security policy is implemented. + var deliveries <-chan tentacle.RemoteSpeechDelivery + if role == "controller" { + runtime.mutex.Lock() + for _, command := range runtime.commands { + if command.id == "remote" { + deliveries = command.session.EnableRemoteSpeech() + break + } + } + runtime.mutex.Unlock() + if deliveries == nil { + t.Fatal("Remote fixture session unavailable") + } + } consent := select_remote_relay(t, forms, listener.Addr().String(), role) credentials := approve_remote_form(t, forms, consent, role) credentials.reply <- internalUIResult{Activated: "inspect-certificate"} @@ -305,6 +334,40 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { if status.form.Title != "Remote relay status" { t.Fatal("missing connection status") } + var delivery tentacle.RemoteSpeechDelivery + if role == "controller" { + select { + case delivery = <-deliveries: + case <-time.After(5 * time.Second): + t.Fatal("split relay speech did not reach the reader owner") + } + var text strings.Builder + ends := 0 + for _, speech := range delivery.Speeches { + if !tentaclewire.ValidRemoteSpeech(speech) || speech.Priority != 2 { + t.Fatal("invalid speech fragment or changed priority") + } + for _, part := range speech.Parts { + if part.End { + ends++ + } else { + if !part.Characters || part.Language != "fr" { + t.Fatal("split speech lost character/language state") + } + text.WriteString(part.Text) + } + } + } + if len(delivery.Speeches) < 2 || text.String() != remoteText || ends != 1 { + t.Fatal("reader owner received partial or altered utterance") + } + select { + case <-deliveries: + t.Fatal("one relay utterance became multiple deliveries") + default: + } + delivery.Release() + } status.reply <- internalUIResult{Activated: "cancel"} select { case <-closed: @@ -312,6 +375,9 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { t.Fatal("Disconnect and close retained the relay") } waitTentacle(t, func() bool { return len(runtime.resultSlots) == 0 }) + if role == "controller" && delivery.Context.Err() == nil { + t.Fatal("Disconnect retained remote delivery authority") + } }) } } diff --git a/internal/tentacle/remote.go b/internal/tentacle/remote.go index 94ba6b4..1025bef 100644 --- a/internal/tentacle/remote.go +++ b/internal/tentacle/remote.go @@ -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 { diff --git a/internal/tentacle/remote_speech.go b/internal/tentacle/remote_speech.go index 8fc791b..637b436 100644 --- a/internal/tentacle/remote_speech.go +++ b/internal/tentacle/remote_speech.go @@ -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. diff --git a/internal/tentacle/remote_speech_test.go b/internal/tentacle/remote_speech_test.go index abb0f86..fd84528 100644 --- a/internal/tentacle/remote_speech_test.go +++ b/internal/tentacle/remote_speech_test.go @@ -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") + } + }) + } +} diff --git a/internal/tentaclewire/remote_speech.go b/internal/tentaclewire/remote_speech.go index 9821610..bc07d1f 100644 --- a/internal/tentaclewire/remote_speech.go +++ b/internal/tentaclewire/remote_speech.go @@ -12,6 +12,7 @@ import ( // Leave room for lease metadata inside the existing 8192-byte call bound. const MaxRemoteSpeechBytes = 6144 const RemoteSpeechQueueSize = 64 +const RemoteSpeechQueueBytes = 256 * 1024 // RemoteSpeechPart carries resolved character/language state for each fragment. // End is an utterance boundary, not text. Whitespace is significant. @@ -31,7 +32,28 @@ type RemoteSpeech struct { type RemotePresentation struct { RemoteOperation - Speech RemoteSpeech `json:"speech"` + Speech RemoteSpeech `json:"speech"` + Utterance uint64 `json:"utterance,omitempty"` // First fragment's operation sequence. + Final bool `json:"final,omitempty"` +} + +// A zero utterance is a complete, single-call effect. A nonzero utterance +// stages speak fragments until Final; the reader admits them as one delivery. +func ValidRemotePresentation(request RemotePresentation) bool { + return len(request.Lease) == 32 && request.Epoch != 0 && request.Sequence != 0 && + ValidRemoteSpeech(request.Speech) && + (request.Utterance == 0 && !request.Final || + request.Utterance > 0 && request.Utterance <= request.Sequence && request.Speech.Kind == "speak") +} + +func (request *RemotePresentation) UnmarshalJSON(data []byte) error { + type plain RemotePresentation + var value plain + if reject_speech_nulls(data) != nil || DecodeStrict(data, &value) != nil || !ValidRemotePresentation(RemotePresentation(value)) { + return ErrProtocol + } + *request = RemotePresentation(value) + return nil } func reject_speech_nulls(data []byte) error { diff --git a/internal/tentaclewire/remote_speech_test.go b/internal/tentaclewire/remote_speech_test.go index c425125..b41c852 100644 --- a/internal/tentaclewire/remote_speech_test.go +++ b/internal/tentaclewire/remote_speech_test.go @@ -90,3 +90,63 @@ func TestRemoteSpeechEncodedBoundary(t *testing.T) { t.Fatal("accepted invalid UTF-8") } } + +func TestRemoteSpeechPresentationSDKValidation(t *testing.T) { + base := `"lease":"` + strings.Repeat("a", 32) + `","epoch":1,"sequence":3,"speech":{"kind":"speak","parts":[{"text":"x"}]}` + cases := []struct { + fields string + valid bool + }{ + {"", true}, + {`,"utterance":3`, true}, + {`,"utterance":1,"final":true`, true}, + {`,"utterance":0,"final":false`, true}, + {`,"utterance":4`, false}, + {`,"utterance":-1`, false}, + {`,"utterance":true`, false}, + {`,"utterance":null`, false}, + {`,"utterance":18446744073709551616`, false}, + {`,"final":true`, false}, + {`,"utterance":1,"final":1`, false}, + {`,"utterance":1,"final":null`, false}, + {`,"extra":1`, false}, + } + var pythonCases []any + for _, item := range cases { + data := []byte(`{` + base + item.fields + `}`) + var request RemotePresentation + if valid := DecodeStrict(data, &request) == nil && ValidRemotePresentation(request); valid != item.valid { + t.Fatalf("Go presentation validation: %s: got %v", item.fields, valid) + } + // Use RawMessage when encoding the cases for Python, preserving uint64 + // boundaries rather than passing them through float64. + pythonCases = append(pythonCases, map[string]any{"request": json.RawMessage(data), "valid": item.valid}) + } + for _, kind := range []string{"cancel", "pause"} { + request := RemotePresentation{RemoteOperation: RemoteOperation{Lease: strings.Repeat("a", 32), Epoch: 1, Sequence: 1}, Utterance: 1, Speech: RemoteSpeech{Kind: kind}} + if kind == "pause" { + paused := true + request.Speech.Paused = &paused + } + if ValidRemotePresentation(request) { + t.Fatal("grouped non-speak effect") + } + pythonCases = append(pythonCases, map[string]any{"request": request, "valid": false}) + } + 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_presentation +for case in json.load(sys.stdin): + assert valid_remote_presentation(case['request']) == case['valid'], str(case)[:180] +` + 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 presentation validation: %v: %s", err, output) + } +} diff --git a/sdk/go/tentacle/session.go b/sdk/go/tentacle/session.go index 1cb26ce..c7cf8da 100644 --- a/sdk/go/tentacle/session.go +++ b/sdk/go/tentacle/session.go @@ -29,6 +29,10 @@ const MaxRemoteSpeechBytes = tentaclewire.MaxRemoteSpeechBytes func ValidRemoteSpeech(speech RemoteSpeech) bool { return tentaclewire.ValidRemoteSpeech(speech) } +func ValidRemotePresentation(request RemotePresentation) bool { + return tentaclewire.ValidRemotePresentation(request) +} + type SessionError struct { Code int `json:"code"` Message string `json:"message"` diff --git a/sdk/python/cthulhu_tentacle/session.py b/sdk/python/cthulhu_tentacle/session.py index aa5f705..806e25d 100644 --- a/sdk/python/cthulhu_tentacle/session.py +++ b/sdk/python/cthulhu_tentacle/session.py @@ -87,6 +87,21 @@ def valid_remote_speech(speech): return False +def valid_remote_presentation(request): + """Validate one speech call; assembly and lease checks belong to core.""" + try: + _fields(request, ("lease", "epoch", "sequence", "speech"), ("utterance", "final")) + utterance, final = request.get("utterance", 0), request.get("final", False) + return (_token(request["lease"]) and _positive(request["epoch"]) + and _positive(request["sequence"]) and valid_remote_speech(request["speech"]) + and type(utterance) is int and 0 <= utterance < 2**64 + and type(final) is bool + and (utterance == 0 and not final + or 0 < utterance <= request["sequence"] and request["speech"]["kind"] == "speak")) + except (ProtocolError, TypeError, ValueError, UnicodeError): + return False + + class SessionError(Exception): def __init__(self, code, message): super().__init__("Tentacle session error " + str(code))