tutti/packages/agent/host/session_actor_test.go

package agenthost

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	storesqlite "github.com/tutti-os/tutti/packages/agent/store-sqlite"
)

func TestSessionActorWaitObservesContextCancellation(t *testing.T) {
	actor := NewSessionActor()
	entered := make(chan struct{})
	release := make(chan struct{})
	done := make(chan error, 1)
	ref := SessionRef{WorkspaceID: "workspace-1", AgentSessionID: "session-1"}
	go func() {
		done <- actor.Do(context.Background(), ref, func(context.Context) error {
			close(entered)
			<-release
			return nil
		})
	}()
	<-entered
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	if err := actor.Do(ctx, ref, func(context.Context) error {
		t.Fatal("canceled waiter entered SessionActor")
		return nil
	}); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("SessionActor.Do() error = %v", err)
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

func TestSendInputRechecksHistoryFenceAfterWaitingForSessionActor(t *testing.T) {
	history := &mutableEffectiveHistory{history: storesqlite.SessionHistory{
		RecoveryState: storesqlite.SessionHistoryRecoveryReady,
	}}
	host := New(Config{
		CanonicalStore:   actorFenceCanonicalStore{},
		Runtime:          actorFenceRuntime{},
		EffectiveHistory: history,
	})
	ref := SessionRef{WorkspaceID: "workspace-1", AgentSessionID: "session-1"}
	entered := make(chan struct{})
	release := make(chan struct{})
	actorDone := make(chan error, 1)
	go func() {
		actorDone <- host.withSessionMutationActor(context.Background(), ref.WorkspaceID, ref.AgentSessionID, func(context.Context) error {
			close(entered)
			<-release
			return nil
		})
	}()
	<-entered
	sendDone := make(chan error, 1)
	go func() {
		_, err := host.SendInput(context.Background(), ref, SendInput{
			Content: []PromptContentBlock{{Type: "text", Text: "ordinary send"}},
		})
		sendDone <- err
	}()
	select {
	case err := <-sendDone:
		t.Fatalf("SendInput bypassed SessionActor: %v", err)
	case <-time.After(20 * time.Millisecond):
	}
	history.setRecoveryState(storesqlite.SessionHistoryRecoveryRollbackPending)
	close(release)
	if err := <-actorDone; err != nil {
		t.Fatal(err)
	}
	if err := <-sendDone; !errors.Is(err, ErrEditRetryInProgress) {
		t.Fatalf("SendInput after history fence error = %v, want ErrEditRetryInProgress", err)
	}
}

type actorFenceCanonicalStore struct{ CanonicalStore }

type actorFenceRuntime struct{ RuntimeController }

type mutableEffectiveHistory struct {
	EffectiveHistoryStore
	mu      sync.Mutex
	history storesqlite.SessionHistory
}

func (s *mutableEffectiveHistory) GetSessionHistory(context.Context, string, string) (storesqlite.SessionHistory, bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.history, true, nil
}

func (s *mutableEffectiveHistory) setRecoveryState(state string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.history.RecoveryState = state
}