tutti/packages/agent/daemon/runtime/controller_session_registry_test.go

package agentruntime

import (
	"context"
	"testing"

	activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
)

type commandEmittingAdapter struct {
	statefulInteractiveAdapter
}

func (*commandEmittingAdapter) Exec(_ context.Context, session Session, _ []PromptContentBlock, _ string, _ string, _ EventSink, emitCommands CommandSnapshotSink) ([]activityshared.Event, error) {
	if emitCommands != nil {
		emitCommands(AgentSessionCommandSnapshot{
			AgentSessionID: session.AgentSessionID,
			Commands: []AgentSessionCommand{{
				Name:        "web",
				Description: "Search the web",
				InputHint:   "query",
			}},
		})
	}
	return nil, nil
}

func TestControllerPublishesAndReplaysCommandSnapshots(t *testing.T) {
	t.Parallel()

	adapter := &statefulInteractiveAdapter{
		commandSnapshot: AgentSessionCommandSnapshot{
			AgentSessionID: "agent-session-1",
			Commands: []AgentSessionCommand{{
				Name:        "init",
				Description: "Initial command",
				InputHint:   "value",
			}},
		},
		hasCommands: true,
	}
	controller := NewController([]Adapter{adapter}, nil)
	started, err := controller.Start(context.Background(), StartInput{
		RoomID:         "room-1",
		AgentSessionID: "agent-session-1",
		Provider:       ProviderCodex,
		Title:          "Test",
	})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}

	events, unsubscribe, ok := controller.Subscribe("room-1", started.Session.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe returned ok=false")
	}
	defer unsubscribe()
	event := waitForStreamEventType(t, events, StreamEventAvailableCommands)
	snapshot, ok := event.Data.(AgentSessionCommandSnapshot)
	if !ok || len(snapshot.Commands) != 1 || snapshot.Commands[0].Name != "init" {
		t.Fatalf("replayed command event = %#v, want initial command snapshot", event)
	}

	liveAdapter := &commandEmittingAdapter{}
	liveController := NewController([]Adapter{liveAdapter}, nil)
	started, err = liveController.Start(context.Background(), StartInput{
		RoomID:         "room-1",
		AgentSessionID: "agent-session-2",
		Provider:       ProviderCodex,
		Title:          "Test",
	})
	if err != nil {
		t.Fatalf("Start live: %v", err)
	}
	liveEvents, liveUnsubscribe, ok := liveController.Subscribe("room-1", started.Session.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe live returned ok=false")
	}
	defer liveUnsubscribe()
	if _, err := liveController.Exec(context.Background(), ExecInput{
		RoomID:         "room-1",
		AgentSessionID: started.Session.AgentSessionID,
		Content:        textPrompt("hello"),
	}); err != nil {
		t.Fatalf("Exec live: %v", err)
	}
	event = waitForStreamEventType(t, liveEvents, StreamEventAvailableCommands)
	snapshot, ok = event.Data.(AgentSessionCommandSnapshot)
	if !ok || len(snapshot.Commands) != 1 || snapshot.Commands[0].Name != "web" {
		t.Fatalf("live command event = %#v, want web command snapshot", event)
	}

	if _, err := liveController.Close(context.Background(), CloseInput{
		RoomID:         "room-1",
		AgentSessionID: started.Session.AgentSessionID,
	}); err != nil {
		t.Fatalf("Close live: %v", err)
	}
	if _, _, ok := liveController.Subscribe("room-1", started.Session.AgentSessionID); ok {
		t.Fatal("Subscribe after close returned ok=true")
	}
}

func TestControllerReplaysCommandSnapshotsQueuedBeforeStartRegistration(t *testing.T) {
	t.Parallel()

	adapter := &statefulInteractiveAdapter{}
	controller := NewController([]Adapter{adapter}, nil)
	if adapter.commandSink == nil {
		t.Fatal("adapter command snapshot sink was not installed")
	}
	adapter.commandSink(AgentSessionCommandSnapshot{
		AgentSessionID: "agent-session-1",
		Commands: []AgentSessionCommand{{
			Name:        "compact",
			Description: "Compact context",
		}},
	})

	started, err := controller.Start(context.Background(), StartInput{
		RoomID:         "room-1",
		AgentSessionID: "agent-session-1",
		Provider:       ProviderCodex,
		Title:          "Test",
	})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}

	events, unsubscribe, ok := controller.Subscribe("room-1", started.Session.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe returned ok=false")
	}
	defer unsubscribe()

	event := waitForStreamEventType(t, events, StreamEventAvailableCommands)
	snapshot, ok := event.Data.(AgentSessionCommandSnapshot)
	if !ok || snapshot.AgentSessionID != "agent-session-1" ||
		len(snapshot.Commands) != 1 ||
		snapshot.Commands[0].Name != "compact" {
		t.Fatalf("replayed command event = %#v, want queued compact command snapshot", event)
	}
}

func TestControllerReplaysCommandSnapshotsQueuedBeforeResumeRegistration(t *testing.T) {
	t.Parallel()

	adapter := &statefulInteractiveAdapter{}
	controller := NewController([]Adapter{adapter}, nil)
	if adapter.commandSink == nil {
		t.Fatal("adapter command snapshot sink was not installed")
	}
	adapter.commandSink(AgentSessionCommandSnapshot{
		AgentSessionID: "agent-session-1",
		Commands: []AgentSessionCommand{{
			Name:        "plan",
			Description: "Enter plan mode",
		}},
	})

	resumed, err := controller.Resume(context.Background(), ResumeInput{
		RoomID:            "room-1",
		AgentSessionID:    "agent-session-1",
		Provider:          ProviderCodex,
		ProviderSessionID: "provider-session-1",
		Title:             "Test",
	})
	if err != nil {
		t.Fatalf("Resume: %v", err)
	}

	events, unsubscribe, ok := controller.Subscribe("room-1", resumed.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe returned ok=false")
	}
	defer unsubscribe()

	event := waitForStreamEventType(t, events, StreamEventAvailableCommands)
	snapshot, ok := event.Data.(AgentSessionCommandSnapshot)
	if !ok || snapshot.AgentSessionID != "agent-session-1" ||
		len(snapshot.Commands) != 1 ||
		snapshot.Commands[0].Name != "plan" {
		t.Fatalf("replayed command event = %#v, want queued plan command snapshot", event)
	}
}

func TestControllerPublishesConfigOptionsUpdatesFromAdapterSink(t *testing.T) {
	t.Parallel()

	adapter := &statefulInteractiveAdapter{}
	controller := NewController([]Adapter{adapter}, nil)
	started, err := controller.Start(context.Background(), StartInput{
		RoomID:         "room-1",
		AgentSessionID: "agent-session-1",
		Provider:       ProviderCodex,
		Title:          "Test",
	})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}
	if adapter.configSink == nil {
		t.Fatal("adapter config options update sink was not installed")
	}

	events, unsubscribe, ok := controller.Subscribe("room-1", started.Session.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe returned ok=false")
	}
	defer unsubscribe()

	adapter.configSink(AgentSessionConfigOptionsUpdate{
		AgentSessionID:  started.Session.AgentSessionID,
		ConfigOptionKey: "model",
	})
	event := waitForStreamEventType(t, events, StreamEventConfigOptions)
	update, ok := event.Data.(AgentSessionConfigOptionsUpdate)
	if !ok {
		t.Fatalf("config options stream event data = %#v, want update payload", event.Data)
	}
	if update.AgentSessionID != started.Session.AgentSessionID ||
		update.Provider != ProviderCodex ||
		update.ProviderSessionID != started.Session.ProviderSessionID ||
		update.ConfigOptionKey != "model" ||
		update.OccurredAtUnixMS <= 0 {
		t.Fatalf("config options update = %#v, want populated payload", update)
	}
}

func TestControllerReplaysConfigOptionsUpdatesQueuedBeforeSessionRegistration(t *testing.T) {
	t.Parallel()

	adapter := &statefulInteractiveAdapter{}
	controller := NewController([]Adapter{adapter}, nil)
	if adapter.configSink == nil {
		t.Fatal("adapter config options update sink was not installed")
	}
	adapter.configSink(AgentSessionConfigOptionsUpdate{
		RoomID:          "room-1",
		AgentSessionID:  "agent-session-1",
		ConfigOptionKey: "model",
	})

	started, err := controller.Start(context.Background(), StartInput{
		RoomID:         "room-1",
		AgentSessionID: "agent-session-1",
		Provider:       ProviderCodex,
		Title:          "Test",
	})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}

	events, unsubscribe, ok := controller.Subscribe("room-1", started.Session.AgentSessionID)
	if !ok {
		t.Fatal("Subscribe returned ok=false")
	}
	defer unsubscribe()

	event := waitForStreamEventType(t, events, StreamEventConfigOptions)
	update, ok := event.Data.(AgentSessionConfigOptionsUpdate)
	if !ok {
		t.Fatalf("config options stream event data = %#v, want update payload", event.Data)
	}
	if update.RoomID != "room-1" ||
		update.AgentSessionID != "agent-session-1" ||
		update.Provider != ProviderCodex ||
		update.ConfigOptionKey != "model" {
		t.Fatalf("config options update = %#v, want replayed model update", update)
	}
}