// Copyright (c) 2026 Storm Dragon // SPDX-License-Identifier: GPL-3.0-or-later package tentacle import ( "context" "encoding/json" "slices" "sync" "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 // Received is the reader's monotonic admission time of the first frame. // Local output can discard old mailbox/assembly work before it is dequeued. Received time.Time // 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 received time.Time speeches []tentaclewire.RemoteSpeech bytes int } // EnableRemoteSpeech installs one bounded mailbox, explicitly from reader code. // Without an owner, remote.present is unavailable. The owner selects session // shutdown as well as this channel and joins its backend cleanup. 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_lease_locked(lease) if access == nil || !slices.Contains(access.grants, grant) { return nil } return access } func (session *Session) remote_lease_locked(lease tentaclewire.RemoteLease) *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 { 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 { 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"} } 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, received: time.Now()} } 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, Received: batch.received, 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 <- delivery: 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"} } }