From b60cf5d5f2408502c31adf70504c8e2cb46db467 Mon Sep 17 00:00:00 2001 From: Angus Kola Date: Wed, 4 Nov 2026 16:53:16 -0500 Subject: [PATCH] Observe authorized local speech for Remote sessions --- docs/tentacle-remote-speech.md | 87 +++- internal/app/event_presentation.go | 1 + internal/app/phonetic_hints.go | 1 + internal/app/speech_delivery.go | 6 + internal/app/speech_interrupt.go | 1 + internal/app/speech_observation.go | 81 ++++ internal/app/speech_observation_test.go | 391 ++++++++++++++++++ internal/app/speech_worker.go | 9 + .../app/tentacle_remote_connection_test.go | 93 ++++- internal/app/tentacles.go | 3 + internal/tentacle/remote_output.go | 123 ++++++ internal/tentacle/remote_output_test.go | 143 +++++++ 12 files changed, 927 insertions(+), 12 deletions(-) create mode 100644 internal/app/speech_observation.go create mode 100644 internal/app/speech_observation_test.go create mode 100644 internal/tentacle/remote_output.go create mode 100644 internal/tentacle/remote_output_test.go diff --git a/docs/tentacle-remote-speech.md b/docs/tentacle-remote-speech.md index 9209410..4d231dd 100644 --- a/docs/tentacle-remote-speech.md +++ b/docs/tentacle-remote-speech.md @@ -1,10 +1,10 @@ # Remote speech transport prerequisite -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 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. +Typed speech IPC now includes an explicit reader-owned incoming mailbox and +limited production observation of ordinary accessibility-event speech. The +external coordinator connects this IPC to relay speech. Incoming speech still +has no production presentation owner and does not make Remote speak. Atomic +assembly of split incoming utterances precedes 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. @@ -72,6 +72,57 @@ there is no child RPC that publishes an observed event. The method queues Full IPC queues fail the process session, cancelling its lease. A receiving Tentacle must recheck lease/epoch and peer readiness before sending over a relay. +## Limited production output observation + +The Tentacle runtime now attaches its manager to the existing speech worker. +`Manager.CaptureRemoteSpeechOutput` returns a reader-only ticket for an opened, +live controlled lease with `output-observe`. The worker captures this ticket +when it admits explicitly local speech. Work queued before consent gets no +ticket; delivery never captures a replacement. The original lease, deadline and +context are checked again when publishing. Production admission also holds the +core availability policy lock, so lock/handoff cannot admit new output after +policy cancellation, including across unlock/return. Already accepted IPC or +relay writes cannot be recalled. + +The zero speech origin is private. Only the main accessibility event presenter +currently marks requests local. Forms, preferences, connection/status messages, +incoming remote text, command-result Tentacle messages and other unclassified +sources stay private. The presenter classifies the original source and ancestors +before context filtering/reset, so a discarded password ancestor still excludes +its descendant's output. Password roles/ancestors, history exclusion, forced mode +announcements, confirmations and protected messages are independently denied. +This does not export every reading path: document/review commands, Say All, +reading-context announcements, separate key-name echo and notification paths +still need explicit source and sensitivity review. Derived phonetic hints retain +their parent's origin and original observation ticket. + +Observation runs after speech-mode suppression, line/indentation processing, +pronunciation, spelling expansion and backend character-path selection. It +records resolved backend **attempts**, before backend I/O, not accepted audio or +audible completion. Character mode is exported only for a single valid CHAR +character; other text uses normal speech. Remote language is the default and +priority is normal; voice, language/profile fidelity and pause/resume remain +unimplemented. No sound cue or backend configuration is sent. + +One local backend chunk becomes an observed utterance. Long UTF-8 text splits +conservatively into at most 64 events, each within the encoded speech limit; +only the last ends the utterance. The complete set of IPC frames must fit the +shared session queue and byte budget, including in-flight bytes, before any +prefix is queued. Invalid/oversized output or overflow retires the process +session, matching existing output-event overflow behavior. Local speech proceeds +independently. This atomic IPC admission is not atomic relay or NVDA delivery. + +User/input interruption and mode/enable changes publish cancel through the same +speech-owner mutex as observed utterances. An older delivery generation cannot +publish behind that cancel. Remote publication never performs pipe, relay or +backend I/O in the input loop. Incoming cancellation ownership remains pending. + +The source command still holds a modal status form for its connection lifetime. +That form captures local keys and suppresses ordinary application presentation; +this increment does not detach it or provide an ordinary desktop workflow. +Keep relay-testing warnings until that UI lifetime and incoming speech ownership +are implemented and live acceptance is recorded. + ## Verification and remaining work Run from this checkout: @@ -91,10 +142,28 @@ 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 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. +The observation increment adds Go/Python SDK tests for captured authority, +UTF-8/escaped splitting, cancel order and stale tickets; isolated host tests for +all-or-none queue admission, in-flight bytes, expired/pending leases and policy +round trips; and speech-worker tests for resolved pronunciation/characters, +chunks/spelling, private origins, modes, generation and lock/handoff retirement. +The source-command/private-relay fixture also exercises production observation +attachment, transformed long output, private exclusion and cancel order with a +test speech backend. It does not drive physical input or audible synthesis. +Queue-level regressions additionally hold the backend across reconnection: +output admitted under old consent or before fresh consent cannot acquire the +replacement lease. An oversized observed utterance retires Remote while both +that local utterance and subsequent local reading still reach the test backend. +Ten repetitions of the observation race cases passed, along with the complete +host/wire/both-SDK race suites (host: about 127 seconds), focused application +speech/Remote/Tentacle race checks, affected-package vet, a separate reader +build, the existing speech regression harness and Tentacles `make check`. + +Next: add incoming speech ownership, delivery-time policy, local priority and +remote-only cancellation, and detach connection status from desktop capture. +Review remaining output producers before assigning exportable origins. +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/event_presentation.go b/internal/app/event_presentation.go index 2c3af7b..2eb7ed7 100644 --- a/internal/app/event_presentation.go +++ b/internal/app/event_presentation.go @@ -235,6 +235,7 @@ func (presenter *eventPresenter) speak(event atspi.Event, capturedIntent ...inpu browseContext = len(context) != 0 } speech.Enqueue(speechRequest{ + origin: accessibility_speech_origin(event.Source), Context: context, ResetContext: !browseContext, VoiceType: ancestor_voice(event.Source.SpeechContext), Window: eventWindow(event), IntroducesWindow: (event.Type == atspi.WindowActivatedEvent || event.Type == atspi.WindowCreateEvent) && diff --git a/internal/app/phonetic_hints.go b/internal/app/phonetic_hints.go index c0ea6c5..35c67de 100644 --- a/internal/app/phonetic_hints.go +++ b/internal/app/phonetic_hints.go @@ -57,6 +57,7 @@ func (worker *speechWorker) deliver_phonetic_hint(ctx context.Context, request s return true, nil } hint := speechRequest{ + origin: request.origin, observation: request.observation, // The character already presented its capital cues. This hint only // clarifies its identity, using the normal voice even in links or Wine. Text: phoneticLetter(letter), VoiceType: "default", Role: request.Role, Cause: "phonetic-character-hint", Append: true, diff --git a/internal/app/speech_delivery.go b/internal/app/speech_delivery.go index 923aa41..770da99 100644 --- a/internal/app/speech_delivery.go +++ b/internal/app/speech_delivery.go @@ -22,6 +22,7 @@ import ( "errors" "fmt" "log/slog" + "strings" "unicode/utf8" ) @@ -137,6 +138,7 @@ func (worker *speechWorker) deliver(ctx context.Context, request speechRequest) return reader.SpeakAndWait(ctx, text, nil) } speak := worker.speaker.Speak + characters := false kind := request.VoiceType if kind == "" { switch request.Role { @@ -156,6 +158,7 @@ func (worker *speechWorker) deliver(ctx context.Context, request speechRequest) SpeakCharacter(context.Context, string) error }); ok { speak = character.SpeakCharacter + characters = true } } if request.CustomEcho && request.Character { @@ -163,6 +166,7 @@ func (worker *speechWorker) deliver(ctx context.Context, request speechRequest) SpeakEchoCharacter(context.Context, string) error }); ok { speak = echo.SpeakEchoCharacter + characters = true } } else if request.CustomEcho { if echo, ok := worker.speaker.(interface { @@ -188,6 +192,7 @@ func (worker *speechWorker) deliver(ctx context.Context, request speechRequest) speak = func(ctx context.Context, text string) error { return voiced.SpeakVoiceAndWait(ctx, text, kind, nil) } + characters = false // This path submits resolved text, not a CHAR command. } } chunks := request.Chunks @@ -223,6 +228,7 @@ func (worker *speechWorker) deliver(ctx context.Context, request speechRequest) part := request part.Append = request.Append || index > 0 worker.record_delivery(ctx, part, text) + worker.observe_speech(ctx, part, text, characters && utf8.RuneCountInString(text) == 1 && !strings.ContainsAny(text, "\r\n\x00")) if err := speak(ctx, text); err != nil { return fmt.Errorf("speak: %w", err) } diff --git a/internal/app/speech_interrupt.go b/internal/app/speech_interrupt.go index 96d4586..fc35b93 100644 --- a/internal/app/speech_interrupt.go +++ b/internal/app/speech_interrupt.go @@ -75,6 +75,7 @@ func (worker *speechWorker) interruptLocked() { worker.pending = nil worker.generation++ worker.interrupting = true + worker.observe_cancel_locked() select { case worker.interrupt <- struct{}{}: diff --git a/internal/app/speech_observation.go b/internal/app/speech_observation.go new file mode 100644 index 0000000..42678b3 --- /dev/null +++ b/internal/app/speech_observation.go @@ -0,0 +1,81 @@ +// Copyright (c) 2026 Storm Dragon +// SPDX-License-Identifier: GPL-3.0-or-later + +package app + +import ( + "context" + + "git.stormux.org/storm/cthulhu/internal/atspi" + "git.stormux.org/storm/cthulhu/internal/tentacle" +) + +// Unclassified output is private. Origins are core metadata, never inferred +// from network text, a Tentacle's request, or a user-visible cause string. +type speechOrigin uint8 + +const ( + speechOriginUnknown speechOrigin = iota + speechOriginLocal + speechOriginRemote + speechOriginStatus + speechOriginSecret +) + +func accessibility_speech_origin(source atspi.Snapshot) speechOrigin { + if source.Role == "password text" { + return speechOriginSecret + } + // Presentation may filter/reset context. Classify the original accessible + // ancestry before doing that, even when its labels will not be spoken. + for _, ancestor := range source.SpeechContext { + if ancestor.Role == "password text" { + return speechOriginSecret + } + } + return speechOriginLocal +} + +func exportable_speech(request speechRequest) bool { + if request.origin != speechOriginLocal || request.Role == "password text" || + request.ExcludeFromHistory || request.Confirmation || request.Uninterruptible || request.ForceSpeech { + return false + } + for _, ancestor := range request.Context { + if ancestor.Role == "password text" { + return false + } + } + return true +} + +func (worker *speechWorker) observe_remote_output(manager *tentacle.Manager) { + worker.mutex.Lock() + defer worker.mutex.Unlock() + worker.remoteOutput = manager.CaptureRemoteSpeechOutput +} + +// Serialize observation with user interruption. No event may be published after +// that interruption's cancel using an older delivery generation. This performs +// bounded IPC queue admission only, never pipe, relay or backend I/O. +func (worker *speechWorker) observe_speech(ctx context.Context, request speechRequest, text string, characters bool) { + if request.observation == nil || !exportable_speech(request) || ctx.Err() != nil { + return + } + worker.mutex.Lock() + defer worker.mutex.Unlock() + if !worker.running || !worker.enabled || worker.secure || worker.generation != request.deliveryGeneration || + worker.mode != speechModeSpeech || ctx.Err() != nil { + return + } + // Failures retire only the remote stream. Local speech remains available. + _ = request.observation.PublishText(text, characters) +} + +func (worker *speechWorker) observe_cancel_locked() { + if worker.remoteOutput != nil && !worker.secure { + if output := worker.remoteOutput(); output != nil { + _ = output.Cancel() + } + } +} diff --git a/internal/app/speech_observation_test.go b/internal/app/speech_observation_test.go new file mode 100644 index 0000000..f8c3ad3 --- /dev/null +++ b/internal/app/speech_observation_test.go @@ -0,0 +1,391 @@ +// Copyright (c) 2026 Storm Dragon +// SPDX-License-Identifier: GPL-3.0-or-later + +package app + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "os" + "strings" + "testing" + "time" + + "git.stormux.org/storm/cthulhu/internal/atspi" + "git.stormux.org/storm/cthulhu/internal/presentation" + "git.stormux.org/storm/cthulhu/internal/tentacle" + "git.stormux.org/storm/cthulhu/internal/tentaclewire" + sdk "git.stormux.org/storm/cthulhu/sdk/go/tentacle" +) + +func TestRemoteSpeechObservationPeer(t *testing.T) { + if os.Args[len(os.Args)-1] != "remote-observation-peer" { + return + } + client := sdk.NewSessionClient() + next := func(ctx context.Context) (sdk.SessionEvent, error) { + select { + case event := <-client.Events(): + return event, nil + case <-ctx.Done(): + return sdk.SessionEvent{}, ctx.Err() + } + } + commands := []sdk.Command{ + {ID: "open", Title: "Open", Handle: func(ctx context.Context, _ json.RawMessage) (any, error) { + event, err := next(ctx) + if err != nil { + return nil, err + } + var lease sdk.RemoteLease + err = client.Call(ctx, "remote.open", sdk.RemoteOpen{Action: event.Action, Epoch: event.Epoch, Role: event.Role, Grants: event.Grants}, &lease) + return lease, err + }}, + {ID: "event", Title: "Event", Handle: func(ctx context.Context, _ json.RawMessage) (any, error) { return next(ctx) }}, + } + if err := sdk.ServeSession(os.Stdin, os.Stdout, commands, nil, client); err != nil { + os.Exit(1) + } + os.Exit(0) +} + +type observationSpeaker struct{ fakeSpeaker } + +func (speaker *observationSpeaker) SpeakCharacter(ctx context.Context, text string) error { + return speaker.Speak(ctx, text) +} + +func observation_fixture(t *testing.T) (*speechWorker, *tentacle.Session, *tentacle.Manager, *tentacle.RemotePolicy) { + t.Helper() + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + manager, err := tentacle.NewManager(tentacle.DefaultLimits(), nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := manager.Shutdown(ctx); err != nil { + t.Error(err) + } + }) + policy := tentacle.NewRemotePolicy() + if err := manager.UseRemotePolicy(policy); err != nil { + t.Fatal(err) + } + session, err := manager.Start(t.Context(), tentacle.Spec{ID: "observation-probe", + Program: []string{executable, "-test.run=^TestRemoteSpeechObservationPeer$", "--", "remote-observation-peer"}, + SessionEvents: true, RemoteGrants: []string{"output-observe"}}) + if err != nil { + t.Fatal(err) + } + select { + case <-session.Ready(): + case <-time.After(5 * time.Second): + t.Fatal("observation peer did not become ready") + } + if err := session.Activate(); err != nil { + t.Fatal(err) + } + approval, err := session.BeginRemoteApproval("controlled", []string{"output-observe"}) + if err != nil { + t.Fatal(err) + } + if _, err := session.AuthorizeRemoteWithApproval(approval, "controlled", []string{"output-observe"}); err != nil { + t.Fatal(err) + } + observation_call(t, session, "open") + speaker := &observationSpeaker{fakeSpeaker{spoken: make(chan string, 64)}} + worker := newSpeechWorker(slog.New(slog.DiscardHandler), speaker) + worker.observe_remote_output(manager) + if err := worker.Start(t.Context()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := worker.Stop(ctx); err != nil { + t.Error(err) + } + }) + return worker, session, manager, policy +} + +func observation_call(t *testing.T, session *tentacle.Session, command string) json.RawMessage { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + result, err := session.Invoke(ctx, command, nil) + if err != nil { + t.Fatal(err) + } + select { + case value := <-result: + if value.Err != nil { + t.Fatal(value.Err) + } + return value.Value + case <-ctx.Done(): + t.Fatal("observation command timed out") + return nil + } +} + +func observed_speech(t *testing.T, session *tentacle.Session) tentaclewire.RemoteSpeech { + t.Helper() + var event tentaclewire.SessionEvent + if err := json.Unmarshal(observation_call(t, session, "event"), &event); err != nil || event.Kind != "remote.output" || event.Speech == nil { + t.Fatal("expected observed speech event", event.Kind, err) + } + return *event.Speech +} + +func TestRemoteSpeechObservationResolvedOutputAndPrivateOrigins(t *testing.T) { + worker, session, _, _ := observation_fixture(t) + worker.textTransform = func(text string) string { return strings.ReplaceAll(text, "colour", "color") } + for _, request := range []speechRequest{ + {Text: "colour", origin: speechOriginLocal}, + {Text: "A", origin: speechOriginLocal, Character: true}, + {Text: "ignored parent", Chunks: []string{"one", "two"}, origin: speechOriginLocal}, + {Spelling: []string{"a", "b"}, Character: true, origin: speechOriginLocal}, + } { + if err := worker.EnqueueAndWait(t.Context(), request); err != nil { + t.Fatal(err) + } + } + for index, text := range []string{"color", "A", "one", "two", "a", "b"} { + speech := observed_speech(t, session) + if speech.Kind != "speak" || len(speech.Parts) != 2 || speech.Parts[0].Text != text || + speech.Parts[0].Characters != (index == 1 || index >= 4) || !speech.Parts[1].End { + t.Fatal("observation preceded policy/character resolution", speech) + } + } + for _, request := range []speechRequest{ + {Text: "unclassified"}, + {Text: "local status", origin: speechOriginStatus}, + {Text: "remote text", origin: speechOriginRemote}, + {Text: "secret", origin: speechOriginSecret}, + {Text: "password", Role: "password text", origin: speechOriginLocal}, + {Text: "private ancestor", Context: []atspi.Snapshot{{Role: "password text"}}, origin: speechOriginLocal}, + {Text: "history", ExcludeFromHistory: true, origin: speechOriginLocal}, + {Text: "mode announcement", ForceSpeech: true, origin: speechOriginLocal}, + {Text: "form value", Cause: "internal-ui"}, + } { + if err := worker.EnqueueAndWait(t.Context(), request); err != nil { + t.Fatal(err) + } + } + worker.InterruptForUser() + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("exported private output before cancellation", speech) + } +} + +func TestRemoteSpeechObservationAccessibilityPresenter(t *testing.T) { + worker, session, _, _ := observation_fixture(t) + state := &eventLoopState{} + state.presentationOptions = presentation.DefaultOptions() + presenter := eventPresenter{state: state, logger: slog.New(slog.DiscardHandler), speech: worker} + presenter.speak(atspi.Event{Type: atspi.CaretEvent, + Source: atspi.Snapshot{Path: "/entry", Role: "entry", CaretCharacter: "a"}}) + speech := observed_speech(t, session) + if speech.Kind != "speak" || speech.Parts[0].Text != "a" || !speech.Parts[0].Characters { + t.Fatal("ordinary accessibility speech did not reach observation", speech) + } + // Context presentation may discard ancestors outside browse mode. Privacy + // classification must still inspect the original accessible ancestry. + presenter.speak(atspi.Event{Type: "focus:", Source: atspi.Snapshot{ + Path: "/nested-entry", Role: "entry", Name: "private descendant", + SpeechContext: []atspi.Snapshot{{Role: "password text"}}, + }}) + if err := worker.EnqueueAndWait(t.Context(), speechRequest{Text: "private completion barrier"}); err != nil { + t.Fatal(err) + } + worker.InterruptForUser() + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("filtered presentation context hid a private ancestor", speech) + } +} + +func TestRemoteSpeechObservationModesAndGeneration(t *testing.T) { + worker, session, manager, _ := observation_fixture(t) + for _, mode := range []speechMode{speechModeOff, speechModeBeeps} { + worker.mutex.Lock() + worker.mode = mode + worker.beep = func(context.Context) error { return nil } + worker.mutex.Unlock() + if err := worker.EnqueueAndWait(t.Context(), speechRequest{Text: "suppressed", origin: speechOriginLocal}); err != nil { + t.Fatal(err) + } + } + worker.mutex.Lock() + worker.mode = speechModeSpeech + old := speechRequest{origin: speechOriginLocal, observation: manager.CaptureRemoteSpeechOutput(), deliveryGeneration: worker.generation} + worker.mutex.Unlock() + worker.InterruptForUser() + worker.observe_speech(t.Context(), old, "stale generation", false) + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("exported suppressed speech", speech) + } + worker.InterruptForUser() + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("older generation overtook cancellation", speech) + } +} + +func TestRemoteSpeechObservationRevokedBeforeDelivery(t *testing.T) { + for _, change := range []string{"revoke", "lock", "handoff"} { + t.Run(change, func(t *testing.T) { + worker, session, manager, policy := observation_fixture(t) + request := speechRequest{origin: speechOriginLocal, observation: manager.CaptureRemoteSpeechOutput()} + if request.observation == nil { + t.Fatal("no observation ticket") + } + switch change { + case "revoke": + session.RevokeRemote() + case "lock": + gate := &focusGate{remotePolicy: policy} + gate.SetLocked(true, nil, nil, nil, worker, &brailleWorker{}) + gate.SetLocked(false, nil, nil, nil, worker, &brailleWorker{}) + case "handoff": + gate := &focusGate{remotePolicy: policy} + gate.SetSuspended(true) + gate.SetSuspended(false) + } + worker.observe_speech(t.Context(), request, "retired", false) + var event tentaclewire.SessionEvent + if err := json.Unmarshal(observation_call(t, session, "event"), &event); err != nil || event.Kind != "remote.revoked" { + t.Fatal("retirement exported text", event.Kind, err) + } + }) + } +} + +type observationBarrierSpeaker struct { + fakeSpeaker + began chan struct{} + release chan struct{} +} + +func (speaker *observationBarrierSpeaker) Speak(ctx context.Context, text string) error { + if text == "old active" { + close(speaker.began) + select { + case <-speaker.release: + case <-ctx.Done(): + return ctx.Err() + } + } + return speaker.fakeSpeaker.Speak(ctx, text) +} + +func TestRemoteSpeechObservationCancellationBarrier(t *testing.T) { + worker, session, _, _ := observation_fixture(t) + speaker := &observationBarrierSpeaker{fakeSpeaker: fakeSpeaker{spoken: make(chan string, 4)}, + began: make(chan struct{}), release: make(chan struct{})} + worker.speaker = speaker // No requests have reached this worker yet. + worker.Enqueue(speechRequest{Text: "old active", origin: speechOriginLocal}) + select { + case <-speaker.began: + case <-time.After(time.Second): + t.Fatal("backend did not block") + } + if speech := observed_speech(t, session); speech.Parts[0].Text != "old active" { + t.Fatal("missing first attempt") + } + worker.Enqueue(speechRequest{Text: "old pending", origin: speechOriginLocal}) + worker.InterruptForUser() + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("cancel overtaken by pending output", speech) + } + close(speaker.release) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + if err := worker.EnqueueAndWait(ctx, speechRequest{Text: "new generation", origin: speechOriginLocal}); err != nil { + t.Fatal(err) + } + if speech := observed_speech(t, session); speech.Parts[0].Text != "new generation" { + t.Fatal("old pending speech crossed interruption", speech) + } +} + +func TestRemoteSpeechObservationQueuedAuthorityDoesNotFollowReconnect(t *testing.T) { + worker, session, _, _ := observation_fixture(t) + speaker := &observationBarrierSpeaker{fakeSpeaker: fakeSpeaker{spoken: make(chan string, 8)}, + began: make(chan struct{}), release: make(chan struct{})} + worker.speaker = speaker + // Hold delivery while output with three distinct authority lifetimes enters + // the real queue. Neither the old lease nor no lease may follow reconnect. + worker.Enqueue(speechRequest{Text: "old active"}) + select { + case <-speaker.began: + case <-time.After(time.Second): + t.Fatal("backend did not block") + } + defer func() { + select { + case <-speaker.release: + default: + close(speaker.release) + } + }() + worker.Enqueue(speechRequest{Text: "old consent", origin: speechOriginLocal}) + session.RevokeRemote() + var event tentaclewire.SessionEvent + if err := json.Unmarshal(observation_call(t, session, "event"), &event); err != nil || event.Kind != "remote.revoked" { + t.Fatal("expected old lease revocation", err) + } + worker.Enqueue(speechRequest{Text: "before consent", origin: speechOriginLocal}) + approval, err := session.BeginRemoteApproval("controlled", []string{"output-observe"}) + if err != nil { + t.Fatal(err) + } + if _, err := session.AuthorizeRemoteWithApproval(approval, "controlled", []string{"output-observe"}); err != nil { + t.Fatal(err) + } + observation_call(t, session, "open") + worker.Enqueue(speechRequest{Text: "new consent", origin: speechOriginLocal}) + close(speaker.release) + if speech := observed_speech(t, session); speech.Kind != "speak" || speech.Parts[0].Text != "new consent" { + t.Fatal("queued output acquired replacement consent", speech) + } + worker.InterruptForUser() + if speech := observed_speech(t, session); speech.Kind != "cancel" { + t.Fatal("unexpected output after replacement delivery", speech) + } +} + +func TestRemoteSpeechObservationOverflowPreservesLocalSpeech(t *testing.T) { + worker, session, _, _ := observation_fixture(t) + // Exceed the complete utterance's frame budget. Remote must retire without + // making local speech depend on a healthy child or exposing a prefix. + text := strings.Repeat("x", 768*tentaclewire.RemoteSpeechQueueSize+1) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + if err := worker.EnqueueAndWait(ctx, speechRequest{Text: text, origin: speechOriginLocal}); err != nil { + t.Fatal("remote overflow interrupted local delivery", err) + } + if state, err := session.Status(); state != tentacle.Failed || !errors.Is(err, tentacle.ErrLimit) { + t.Fatal("remote overflow retained a healthy session", state, err) + } + if err := worker.EnqueueAndWait(ctx, speechRequest{Text: "local reading continues", origin: speechOriginLocal}); err != nil { + t.Fatal("retired remote session blocked local reading", err) + } + speaker := worker.speaker.(*observationSpeaker) + for _, expected := range []string{text, "local reading continues"} { + select { + case actual := <-speaker.spoken: + if actual != expected { + t.Fatal("remote failure changed local speech") + } + default: + t.Fatal("missing local speech after remote overflow") + } + } +} diff --git a/internal/app/speech_worker.go b/internal/app/speech_worker.go index a57fcb8..ab2a86b 100644 --- a/internal/app/speech_worker.go +++ b/internal/app/speech_worker.go @@ -30,6 +30,7 @@ import ( "git.stormux.org/storm/cthulhu/internal/config" "git.stormux.org/storm/cthulhu/internal/input" "git.stormux.org/storm/cthulhu/internal/sound" + "git.stormux.org/storm/cthulhu/internal/tentacle" ) // speechQueueCapacity bounds work retained while the speech backend is slow. @@ -38,6 +39,9 @@ import ( const speechQueueCapacity = 64 type speechRequest struct { + origin speechOrigin + observation *tentacle.RemoteSpeechOutput // Captured at admission, never refreshed during delivery. + presentationGeneration *uint64 // Checked under the speech owner mutex. presentationRevision *uint64 @@ -86,6 +90,7 @@ type readAllPart struct { // have separate goroutines and joins; activeDone and generation form their // barrier so cancellation cannot overtake newly accepted output. type speechWorker struct { + remoteOutput func() *tentacle.RemoteSpeechOutput // Owned under mutex; no runtime/output locks in callback. logger *slog.Logger speaker Speaker history *speechHistory @@ -197,6 +202,10 @@ func (worker *speechWorker) Enqueue(request speechRequest) { completeSpeechRequest(request, errors.New("speech worker is not running")) return } + request.observation = nil + if worker.remoteOutput != nil && exportable_speech(request) && !worker.secure { + request.observation = worker.remoteOutput() + } request.Append = true // A Tentacle ticket gates admission, not the lifetime of accepted speech. // Passive focus changes can retire the ticket while this output finishes. diff --git a/internal/app/tentacle_remote_connection_test.go b/internal/app/tentacle_remote_connection_test.go index e3dd6ec..dd7037c 100644 --- a/internal/app/tentacle_remote_connection_test.go +++ b/internal/app/tentacle_remote_connection_test.go @@ -240,6 +240,7 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { } defer listener.Close() probed, joined, closed := make(chan struct{}), make(chan struct{}), make(chan struct{}) + relayOutput := make(chan []byte, 64) go func() { probe, err := listener.Accept() if err != nil { @@ -289,12 +290,33 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { }}) } close(joined) - io.Copy(io.Discard, reader) + for { + frame, err := reader.ReadBytes('\n') + if err != nil { + break + } + relayOutput <- frame + } 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. + if role == "controlled" { + worker := runtime.output.speech + worker.speaker = &fakeSpeaker{spoken: make(chan string, 16)} + worker.textTransform = func(text string) string { return strings.ReplaceAll(text, "colour", "color") } + if err := worker.Start(t.Context()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := worker.Stop(ctx); err != nil { + t.Error(err) + } + }) + } + // Incoming speech still needs a test owner; production observation + // is attached through makeManager on the controlled side. var deliveries <-chan tentacle.RemoteSpeechDelivery if role == "controller" { runtime.mutex.Lock() @@ -367,6 +389,71 @@ func TestRemoteConnectionCommandPrivateRelay(t *testing.T) { default: } delivery.Release() + } else { + // Joining at the relay is earlier than the Tentacle admitting the + // peer. Poll status before testing actual output rather than + // depending on a scheduling delay or replaying pre-peer output. + deadline := time.Now().Add(5 * time.Second) + for { + encoded, _ := json.Marshal(status.form) + if strings.Contains(string(encoded), "One opposite-role peer has joined") { + break + } + if time.Now().After(deadline) { + t.Fatal("peer did not become ready") + } + status.reply <- internalUIResult{Activated: "refresh"} + status = next_remote_form(t, forms) + } + worker := runtime.output.speech + for _, request := range []speechRequest{ + {Text: "private form", Cause: "internal-ui"}, + {Text: "password", Role: "password text", origin: speechOriginLocal}, + {Text: "local status", origin: speechOriginStatus}, + {Text: "colour " + remoteText, origin: speechOriginLocal}, + } { + if err := worker.EnqueueAndWait(t.Context(), request); err != nil { + t.Fatal(err) + } + } + worker.InterruptForUser() + var received strings.Builder + ends, frames := 0, 0 + readOutput: + for { + select { + case frame := <-relayOutput: + var message struct { + Type string `json:"type"` + Sequence []json.RawMessage `json:"sequence"` + } + if err := json.Unmarshal(frame, &message); err != nil { + t.Fatal(err) + } + if message.Type == "cancel" { + break readOutput + } + if message.Type != "speak" { + t.Fatal("unexpected observed relay message", message.Type) + } + frames++ + for _, raw := range message.Sequence { + var text string + if json.Unmarshal(raw, &text) == nil { + received.WriteString(text) + } else if strings.Contains(string(raw), "EndUtteranceCommand") { + ends++ + } else { + t.Fatal("unexpected observed speech command", string(raw)) + } + } + case <-time.After(5 * time.Second): + t.Fatal("reader output did not reach the relay in order") + } + } + if received.String() != "color "+remoteText || frames < 2 || ends != 1 { + t.Fatal("observation leaked private text or lost resolved output/boundaries") + } } status.reply <- internalUIResult{Activated: "cancel"} select { diff --git a/internal/app/tentacles.go b/internal/app/tentacles.go index 4f2a81f..2878596 100644 --- a/internal/app/tentacles.go +++ b/internal/app/tentacles.go @@ -426,6 +426,9 @@ func (runtime *tentacleRuntime) makeManager() (*tentacle.Manager, error) { if err == nil && runtime.remotePolicy != nil { err = manager.UseRemotePolicy(runtime.remotePolicy) } + if err == nil && runtime.output != nil && runtime.output.speech != nil { + runtime.output.speech.observe_remote_output(manager) + } return manager, err } diff --git a/internal/tentacle/remote_output.go b/internal/tentacle/remote_output.go new file mode 100644 index 0000000..a461b42 --- /dev/null +++ b/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 +} diff --git a/internal/tentacle/remote_output_test.go b/internal/tentacle/remote_output_test.go new file mode 100644 index 0000000..626346d --- /dev/null +++ b/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") + } + }) + } +}