124 lines
4.1 KiB
Go
124 lines
4.1 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package tentacle
|
|
|
|
import (
|
|
"context"
|
|
"unicode/utf8"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
|
|
)
|
|
|
|
// RemoteSpeechOutput is a reader-only observation ticket. Capture it when local
|
|
// output is admitted, then retain it through delivery. It cannot follow a lease
|
|
// replacement or authorize output queued before consent. No child RPC exposes it.
|
|
type RemoteSpeechOutput struct {
|
|
session *Session
|
|
lease tentaclewire.RemoteLease
|
|
ctx context.Context
|
|
}
|
|
|
|
func (manager *Manager) CaptureRemoteSpeechOutput() *RemoteSpeechOutput {
|
|
manager.remoteMutex.Lock()
|
|
session := manager.remoteOwner
|
|
manager.remoteMutex.Unlock()
|
|
if session == nil {
|
|
return nil
|
|
}
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
access := session.remote
|
|
if access == nil {
|
|
return nil
|
|
}
|
|
lease := tentaclewire.RemoteLease{Lease: access.lease, Epoch: access.epoch}
|
|
if session.remote_access_locked(lease, "output-observe") == nil || access.role != "controlled" {
|
|
return nil
|
|
}
|
|
return &RemoteSpeechOutput{session: session, lease: lease, ctx: access.ctx}
|
|
}
|
|
|
|
// PublishText exports one resolved backend utterance. The caller must establish
|
|
// local origin and exclude sensitive/status output. Conservative UTF-8 chunks
|
|
// fit even six-byte JSON escaping; only the final frame ends the utterance.
|
|
// The complete set is reserved before any frame becomes visible to the child.
|
|
func (output *RemoteSpeechOutput) PublishText(text string, characters bool) error {
|
|
if output == nil || output.ctx.Err() != nil {
|
|
return ErrPermission
|
|
}
|
|
const chunkBytes = 768
|
|
if text == "" {
|
|
return nil
|
|
}
|
|
if len(text) > tentaclewire.RemoteSpeechQueueBytes || !utf8.ValidString(text) {
|
|
return output.publish(nil, true)
|
|
}
|
|
var speeches []tentaclewire.RemoteSpeech
|
|
for len(text) != 0 {
|
|
if len(speeches) == tentaclewire.RemoteSpeechQueueSize {
|
|
return output.publish(nil, true)
|
|
}
|
|
end := min(len(text), chunkBytes)
|
|
for end < len(text) && !utf8.RuneStart(text[end]) {
|
|
end--
|
|
}
|
|
parts := []tentaclewire.RemoteSpeechPart{{Text: text[:end], Characters: characters}}
|
|
text = text[end:]
|
|
if text == "" {
|
|
parts = append(parts, tentaclewire.RemoteSpeechPart{End: true})
|
|
}
|
|
speeches = append(speeches, tentaclewire.RemoteSpeech{Kind: "speak", Parts: parts})
|
|
}
|
|
return output.publish(speeches, false)
|
|
}
|
|
|
|
func (output *RemoteSpeechOutput) Cancel() error {
|
|
if output == nil {
|
|
return ErrPermission
|
|
}
|
|
return output.publish([]tentaclewire.RemoteSpeech{{Kind: "cancel"}}, false)
|
|
}
|
|
|
|
func (output *RemoteSpeechOutput) publish(speeches []tentaclewire.RemoteSpeech, overflow bool) error {
|
|
session := output.session
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
// Serialize production admission with lock/handoff cancellation. Policy
|
|
// cancellation never takes a session lock, so this follows consent's order.
|
|
if policy := session.manager.remotePolicy; policy != nil {
|
|
policy.mutex.Lock()
|
|
defer policy.mutex.Unlock()
|
|
}
|
|
access := session.remote_access_locked(output.lease, "output-observe")
|
|
if access == nil || access.role != "controlled" || output.ctx.Err() != nil {
|
|
return ErrPermission
|
|
}
|
|
frames := make([]outgoing, 0, len(speeches))
|
|
bytes := 0
|
|
for index, speech := range speeches {
|
|
event := tentaclewire.SessionEvent{Sequence: session.eventSequence + uint64(index) + 1,
|
|
Kind: "remote.output", Lease: output.lease.Lease, Epoch: output.lease.Epoch, Speech: &speech}
|
|
frame, err := encode_request("", "session.event", event)
|
|
if err != nil || !tentaclewire.ValidRemoteSpeech(speech) {
|
|
overflow = true
|
|
break
|
|
}
|
|
frames = append(frames, outgoing{frame: frame})
|
|
bytes += len(frame)
|
|
}
|
|
if overflow || len(frames) > cap(session.outbound)-len(session.outbound) ||
|
|
session.queuedBytes+bytes > tentaclewire.SessionQueueBytes {
|
|
// Losing a cancel or a suffix must never leave an apparently healthy
|
|
// speech stream. Match the existing observed-output overflow policy.
|
|
session.fail_locked(ErrLimit)
|
|
return ErrLimit
|
|
}
|
|
for _, frame := range frames {
|
|
session.outbound <- frame // All producers reserve under session.mutex.
|
|
session.queuedBytes += len(frame.frame)
|
|
}
|
|
session.eventSequence += uint64(len(frames))
|
|
return nil
|
|
}
|