cthulhu/internal/tentacle/remote_speech.go

// 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"}
	}
}