404 lines
15 KiB
Go
404 lines
15 KiB
Go
// 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/config"
|
|
"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"}},
|
|
}})
|
|
for _, source := range []atspi.Snapshot{
|
|
{Role: "unknown"}, {Role: "invalid"}, {Role: "entry", Defunct: true},
|
|
{Role: "entry", SpeechContext: []atspi.Snapshot{{Role: "unknown"}}},
|
|
{Role: "entry", SpeechContext: []atspi.Snapshot{{Role: "panel", Defunct: true}}},
|
|
} {
|
|
source.Path, source.Name = "/uncertain-entry", "unclassified source"
|
|
presenter.speak(atspi.Event{Type: "focus:", Source: source})
|
|
}
|
|
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)
|
|
worker.reading = func() config.ReadingSettings { return config.DefaultSettings().Reading }
|
|
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)
|
|
}
|
|
}()
|
|
list := atspi.Snapshot{BusName: ":1.42", Path: "/list", Role: "list", Name: "Old context"}
|
|
worker.Enqueue(speechRequest{Text: "old consent", origin: speechOriginLocal, Context: []atspi.Snapshot{list}})
|
|
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)
|
|
}
|
|
list.Name = "Before consent"
|
|
worker.Enqueue(speechRequest{Text: "before consent", origin: speechOriginLocal, Context: []atspi.Snapshot{list}})
|
|
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, Context: []atspi.Snapshot{}})
|
|
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")
|
|
}
|
|
}
|
|
}
|