Observe authorized local speech for Remote sessions

internal/tentacle/remote_output.go

@@ -0,0 +1,123 @@
+// 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
+}

internal/tentacle/remote_output_test.go

@@ -0,0 +1,143 @@
+// Copyright (c) 2026 Storm Dragon
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package tentacle
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.stormux.org/storm/cthulhu/internal/tentaclewire"
+)
+
+func TestRemoteSpeechOutputSDKTicket(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")
+			if manager.CaptureRemoteSpeechOutput() != nil {
+				t.Fatal("captured output without consent")
+			}
+			lease := speech_fixture_lease(t, session, "controller", []string{"remote-present"})
+			if manager.CaptureRemoteSpeechOutput() != nil {
+				t.Fatal("controller acquired observation")
+			}
+			session.RevokeRemote()
+			invokeSessionSDK(t, session, "event", nil) // Revocation.
+			lease = speech_fixture_lease(t, session, "controlled", []string{"output-observe"})
+			output := manager.CaptureRemoteSpeechOutput()
+			if output == nil {
+				t.Fatal("live controlled output unavailable")
+			}
+			text := "  " + strings.Repeat("é\x00世界\t", 170) + "  "
+			if err := output.PublishText(text, true); err != nil {
+				t.Fatal(err)
+			}
+			var received strings.Builder
+			frames, ends := 0, 0
+			for ends == 0 {
+				var event tentaclewire.SessionEvent
+				if err := json.Unmarshal(invokeSessionSDK(t, session, "event", nil), &event); err != nil {
+					t.Fatal(err)
+				}
+				frames++
+				if event.Kind != "remote.output" || event.Lease != lease.Lease || event.Epoch != lease.Epoch ||
+					event.Speech == nil || !tentaclewire.ValidRemoteSpeech(*event.Speech) {
+					t.Fatal("invalid observed event", event)
+				}
+				for _, part := range event.Speech.Parts {
+					if part.End {
+						ends++
+					} else {
+						if !part.Characters || part.Language != "" {
+							t.Fatal("changed character/default-language state")
+						}
+						received.WriteString(part.Text)
+					}
+				}
+			}
+			if frames < 2 || ends != 1 || received.String() != text {
+				t.Fatal("changed split text or boundaries")
+			}
+			if err := output.Cancel(); err != nil {
+				t.Fatal(err)
+			}
+			var event tentaclewire.SessionEvent
+			if json.Unmarshal(invokeSessionSDK(t, session, "event", nil), &event) != nil || event.Speech.Kind != "cancel" {
+				t.Fatal("lost cancellation order")
+			}
+			session.RevokeRemote()
+			invokeSessionSDK(t, session, "event", nil)
+			_ = speech_fixture_lease(t, session, "controlled", []string{"output-observe"})
+			if !errors.Is(output.PublishText("old", false), ErrPermission) || !errors.Is(output.Cancel(), ErrPermission) {
+				t.Fatal("old ticket followed replacement lease")
+			}
+		})
+	}
+}
+
+// No pipe writer: prove that a split output reserves all queue space before
+// publishing a prefix, including encoded-byte and in-flight accounting.
+func TestRemoteSpeechOutputAdmissionAndPolicy(t *testing.T) {
+	for _, mode := range []string{"slots", "bytes", "frames", "invalid", "policy", "expired", "pending"} {
+		t.Run(mode, func(t *testing.T) {
+			manager := test_manager(t, DefaultLimits(), nil)
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			outputContext, revoke := context.WithCancel(ctx)
+			defer revoke()
+			session := &Session{manager: manager, spec: Spec{SessionEvents: true}, ctx: ctx, cancel: cancel,
+				outputContext: outputContext, revokeOutput: revoke, state: Active,
+				outbound: make(chan outgoing, 2), revoked: make(chan struct{})}
+			accessContext, retire := context.WithCancel(ctx)
+			defer retire()
+			session.remote = &remoteAccess{lease: strings.Repeat("a", 32), epoch: 1, role: "controlled",
+				grants: []string{"output-observe"}, ctx: accessContext, cancel: retire, deadline: time.Now().Add(time.Minute)}
+			manager.remoteOwner = session
+			output := manager.CaptureRemoteSpeechOutput()
+			text := strings.Repeat("x", 800)
+			want := ErrLimit
+			switch mode {
+			case "slots":
+				session.outbound <- outgoing{frame: []byte("existing")}
+			case "bytes":
+				session.queuedBytes = tentaclewire.SessionQueueBytes - 1
+			case "frames":
+				text = strings.Repeat("x", 768*tentaclewire.RemoteSpeechQueueSize+1)
+			case "invalid":
+				text = "\xff"
+			case "policy":
+				manager.remotePolicy = NewRemotePolicy()
+				session.remote.ctx, session.remote.cancel = context.WithCancel(manager.remotePolicy.Context())
+				output.ctx = session.remote.ctx
+				defer session.remote.cancel()
+				manager.remotePolicy.SetAvailable(false)
+				manager.remotePolicy.SetAvailable(true)
+				want = ErrPermission
+			case "expired":
+				session.remote.deadline = time.Now().Add(-time.Second)
+				want = ErrPermission
+			case "pending":
+				session.remote.lease = ""
+				want = ErrPermission
+			}
+			before := len(session.outbound)
+			if err := output.PublishText(text, false); !errors.Is(err, want) {
+				t.Fatalf("publish = %v, want %v", err, want)
+			}
+			if len(session.outbound) != before || session.eventSequence != 0 {
+				t.Fatal("failed admission published a prefix or consumed event sequence")
+			}
+			if want == ErrLimit && (session.state != Failed || accessContext.Err() == nil) {
+				t.Fatal("overflow retained live stream")
+			}
+			if want == ErrPermission && manager.CaptureRemoteSpeechOutput() != nil {
+				t.Fatal("captured invalid output authority")
+			}
+		})
+	}
+}