iop/apps/node/internal/adapters/cli/persistent/execute_test.go

package persistent_test

import (
	"context"
	"strings"
	"testing"
	"time"

	"go.uber.org/zap"

	clipkg "iop/apps/node/internal/adapters/cli"
	"iop/apps/node/internal/adapters/cli/internal/testutil"
	noderuntime "iop/apps/node/internal/runtime"
	"iop/packages/config"
)

func TestCLIExecutePersistentWaitsForSlowFirstOutput(t *testing.T) {
	testutil.RequirePTYSupport(t)

	cfg := config.CLIConf{
		Enabled: true,
		Profiles: map[string]config.CLIProfileConf{
			"slow-echo": {
				Command:               "sh",
				Args:                  []string{"-c", `stty -echo; while IFS= read -r line; do sleep 0.2; printf "reply:%s\n" "$line"; done`},
				Persistent:            true,
				Terminal:              true,
				ResponseIdleTimeoutMS: 50,
				StartupIdleTimeoutMS:  50,
			},
		},
	}
	c := clipkg.New(cfg, zap.NewNop())

	ctx := context.Background()
	if err := c.Start(ctx); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer func() { _ = c.Stop(ctx) }()

	execCtx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()

	sink := &testutil.FakeSink{}
	err := c.Execute(execCtx, noderuntime.ExecutionSpec{
		RunID: "run-1",
		Model: "slow-echo",
		Input: map[string]any{"prompt": "hello"},
	}, sink)
	if err != nil {
		t.Fatalf("execute: %v", err)
	}

	events := sink.Events()
	if events[len(events)-1].Type != noderuntime.EventTypeComplete {
		t.Fatalf("expected last event to be complete, got %q", events[len(events)-1].Type)
	}
	if combined := testutil.CollectDeltas(events); !strings.Contains(combined, "reply:hello") {
		t.Fatalf("expected reply:hello in deltas, got %q", combined)
	}
}

func TestCLIExecutePersistentProcessExitReturnsError(t *testing.T) {
	testutil.RequirePTYSupport(t)

	cfg := config.CLIConf{
		Enabled: true,
		Profiles: map[string]config.CLIProfileConf{
			"exit-on-input": {
				Command:               "sh",
				Args:                  []string{"-c", `stty -echo; while IFS= read -r line; do printf "before-exit\n"; exit 2; done`},
				Persistent:            true,
				Terminal:              true,
				ResponseIdleTimeoutMS: 500,
				StartupIdleTimeoutMS:  50,
			},
		},
	}
	c := clipkg.New(cfg, zap.NewNop())

	ctx := context.Background()
	if err := c.Start(ctx); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer func() { _ = c.Stop(ctx) }()

	sink := &testutil.FakeSink{}
	err := c.Execute(ctx, noderuntime.ExecutionSpec{
		RunID: "run-1",
		Model: "exit-on-input",
		Input: map[string]any{"prompt": "hello"},
	}, sink)
	if err == nil {
		t.Fatal("expected non-nil error when persistent process exits")
	}

	events := sink.Events()
	var hasError bool
	for _, e := range events {
		if e.Type == noderuntime.EventTypeError {
			hasError = true
		}
	}
	if !hasError {
		t.Fatal("expected error event in emitted events")
	}
	if len(events) > 0 && events[len(events)-1].Type == noderuntime.EventTypeComplete {
		t.Fatal("last event should not be complete when process exits unexpectedly")
	}
}

func TestCLIStartPersistentTerminalAndExecuteWritesPrompt(t *testing.T) {
	testutil.RequirePTYSupport(t)

	cfg := config.CLIConf{
		Enabled: true,
		Profiles: map[string]config.CLIProfileConf{
			"echo-persistent": {
				Command:               "sh",
				Args:                  []string{"-c", `stty -echo; while IFS= read -r line; do printf "reply:%s\n" "$line"; done`},
				Persistent:            true,
				Terminal:              true,
				ResponseIdleTimeoutMS: 300,
				StartupIdleTimeoutMS:  50,
			},
		},
	}
	c := clipkg.New(cfg, zap.NewNop())

	ctx := context.Background()
	if err := c.Start(ctx); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer func() { _ = c.Stop(ctx) }()

	sink := &testutil.FakeSink{}
	err := c.Execute(ctx, noderuntime.ExecutionSpec{
		RunID: "run-1",
		Model: "echo-persistent",
		Input: map[string]any{"prompt": "hello"},
	}, sink)
	if err != nil {
		t.Fatalf("execute: %v", err)
	}

	events := sink.Events()
	var started, completed bool
	for _, e := range events {
		switch e.Type {
		case noderuntime.EventTypeStart:
			started = true
		case noderuntime.EventTypeComplete:
			completed = true
		}
	}
	if !started {
		t.Fatal("expected start event")
	}
	if !completed {
		t.Fatal("expected complete event")
	}
	if combined := testutil.CollectDeltas(events); !strings.Contains(combined, "reply:hello") {
		t.Fatalf("expected reply:hello in deltas, got %q", combined)
	}
}

func TestCLIExecutePersistentTimeoutCleansSession(t *testing.T) {
	testutil.RequirePTYSupport(t)

	cfg := config.CLIConf{
		Enabled: true,
		Profiles: map[string]config.CLIProfileConf{
			"slow-echo": {
				Command:               "sh",
				Args:                  []string{"-c", `stty -echo; while IFS= read -r line; do sleep 0.1; printf "reply:%s\n" "$line"; done`},
				Persistent:            true,
				Terminal:              true,
				ResponseIdleTimeoutMS: 5000,
				StartupIdleTimeoutMS:  50,
			},
		},
	}
	c := clipkg.New(cfg, zap.NewNop())

	ctx := context.Background()
	if err := c.Start(ctx); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer func() { _ = c.Stop(ctx) }()

	firstCtx, firstCancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer firstCancel()

	sink1 := &testutil.FakeSink{}
	err := c.Execute(firstCtx, noderuntime.ExecutionSpec{
		RunID: "run-1",
		Model: "slow-echo",
		Input: map[string]any{"prompt": "first"},
	}, sink1)
	if err == nil {
		t.Fatal("expected context error on first timeout execute")
	}

	time.Sleep(300 * time.Millisecond)

	sink2 := &testutil.FakeSink{}
	err = c.Execute(ctx, noderuntime.ExecutionSpec{
		RunID: "run-2",
		Model: "slow-echo",
		Input: map[string]any{"prompt": "second"},
	}, sink2)
	if err == nil {
		t.Fatal("expected no-session error on second execute after timeout")
	}

	for _, e := range sink2.Events() {
		if e.Type == noderuntime.EventTypeDelta {
			t.Fatalf("unexpected delta event in second run (session contamination): %q", e.Delta)
		}
	}
}

func TestCLIExecutePersistentConsecutiveExecutes(t *testing.T) {
	testutil.RequirePTYSupport(t)

	cfg := config.CLIConf{
		Enabled: true,
		Profiles: map[string]config.CLIProfileConf{
			"consecutive-echo": {
				Command:               "sh",
				Args:                  []string{"-c", `stty -echo; while IFS= read -r line; do printf "reply:%s\n" "$line"; done`},
				Persistent:            true,
				Terminal:              true,
				ResponseIdleTimeoutMS: 100,
				StartupIdleTimeoutMS:  50,
			},
		},
	}
	c := clipkg.New(cfg, zap.NewNop())

	ctx := context.Background()
	if err := c.Start(ctx); err != nil {
		t.Fatalf("start: %v", err)
	}
	defer func() { _ = c.Stop(ctx) }()

	sink1 := &testutil.FakeSink{}
	err := c.Execute(ctx, noderuntime.ExecutionSpec{
		RunID: "run-1",
		Model: "consecutive-echo",
		Input: map[string]any{"prompt": "first"},
	}, sink1)
	if err != nil {
		t.Fatalf("first execute: %v", err)
	}
	events1 := sink1.Events()
	if events1[len(events1)-1].Type != noderuntime.EventTypeComplete {
		t.Fatalf("first run: expected last event to be complete, got %q", events1[len(events1)-1].Type)
	}

	sink2 := &testutil.FakeSink{}
	err = c.Execute(ctx, noderuntime.ExecutionSpec{
		RunID: "run-2",
		Model: "consecutive-echo",
		Input: map[string]any{"prompt": "second"},
	}, sink2)
	if err != nil {
		t.Fatalf("second execute: %v", err)
	}
	events2 := sink2.Events()
	if events2[len(events2)-1].Type != noderuntime.EventTypeComplete {
		t.Fatalf("second run: expected last event to be complete, got %q", events2[len(events2)-1].Type)
	}

	for _, e := range events2 {
		if e.Type == noderuntime.EventTypeDelta && strings.Contains(e.Delta, "first") {
			t.Fatalf("second run received first run's delta (session contamination): %q", e.Delta)
		}
	}
}