package cli_test import ( "context" "fmt" "os" "path/filepath" "runtime" "strings" "sync" "testing" "time" "go.uber.org/zap" noderuntime "iop/apps/node/internal/runtime" "iop/packages/config" . "iop/apps/node/internal/adapters/cli" ) type fakeSink struct { mu sync.Mutex events []noderuntime.RuntimeEvent } func (f *fakeSink) Emit(_ context.Context, e noderuntime.RuntimeEvent) error { f.mu.Lock() defer f.mu.Unlock() f.events = append(f.events, e) return nil } func (f *fakeSink) Events() []noderuntime.RuntimeEvent { f.mu.Lock() defer f.mu.Unlock() cp := make([]noderuntime.RuntimeEvent, len(f.events)) copy(cp, f.events) return cp } func TestCLIExecuteOneShotPassesPromptAsArg(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("sh not available on Windows") } cfg := config.CLIConf{ Enabled: true, Profiles: map[string]config.CLIProfileConf{ "echo": { Command: "sh", // $0 = "sh" (script name), $1 = prompt appended by executeOneShot Args: []string{"-c", `printf "reply:%s\n" "$1"`, "sh"}, }, }, } c := New(cfg, zap.NewNop()) sink := &fakeSink{} err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ RunID: "run-1", Model: "echo", Input: map[string]any{"prompt": "hello"}, }, sink) if err != nil { t.Fatalf("execute: %v", err) } events := sink.Events() var types []string var deltas []string for _, e := range events { types = append(types, string(e.Type)) if e.Type == noderuntime.EventTypeDelta { deltas = append(deltas, e.Delta) } } if len(events) < 3 { t.Fatalf("expected at least 3 events (start/delta/complete), got %v", types) } if events[0].Type != noderuntime.EventTypeStart { t.Fatalf("first event should be start, got %q", events[0].Type) } if events[len(events)-1].Type != noderuntime.EventTypeComplete { t.Fatalf("last event should be complete, got %q", events[len(events)-1].Type) } combined := strings.Join(deltas, "") if !strings.Contains(combined, "reply:hello") { t.Fatalf("expected reply:hello in deltas, got %q", combined) } } func TestCLIExecutePersistentWaitsForSlowFirstOutput(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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, // shorter than the 200ms sleep StartupIdleTimeoutMS: 50, }, }, } c := 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 := &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() var deltas []string var lastType noderuntime.EventType for _, e := range events { if e.Type == noderuntime.EventTypeDelta { deltas = append(deltas, e.Delta) } lastType = e.Type } if lastType != noderuntime.EventTypeComplete { t.Fatalf("expected last event to be complete, got %q", lastType) } combined := strings.Join(deltas, "") if !strings.Contains(combined, "reply:hello") { t.Fatalf("expected reply:hello in deltas, got %q", combined) } } func TestCLIExecutePersistentProcessExitReturnsError(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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 := 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 := &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 TestCLIStartPersistentReturnsErrorWhenProcessExitsDuringStartup(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } cfg := config.CLIConf{ Enabled: true, Profiles: map[string]config.CLIProfileConf{ "exits-immediately": { Command: "sh", Args: []string{"-c", "exit 2"}, Persistent: true, Terminal: true, StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() err := c.Start(ctx) if err == nil { t.Fatal("expected error when persistent process exits during startup") } } func TestCLIStartCleansUpStartedProfilesWhenLaterProfileFails(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("Unix shell required") } cfg := config.CLIConf{ Enabled: true, Profiles: map[string]config.CLIProfileConf{ "ok-profile": { Command: "sh", Args: []string{"-c", `while IFS= read -r line; do printf "reply:%s\n" "$line"; done`}, Persistent: true, Terminal: false, StartupIdleTimeoutMS: 50, ResponseIdleTimeoutMS: 200, }, "fail-profile": { Command: "sh", Args: []string{"-c", "exit 2"}, Persistent: true, Terminal: true, StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() if err := c.Start(ctx); err == nil { t.Fatal("expected Start to fail when one persistent profile exits immediately") } // Stop must not panic after partial start failure. _ = c.Stop(ctx) // Second Stop must also not panic (idempotency). _ = c.Stop(ctx) // Sessions must be cleared — Execute should return a no-session error for every profile. for _, model := range []string{"ok-profile", "fail-profile"} { sink := &fakeSink{} if err := c.Execute(ctx, noderuntime.ExecutionSpec{ RunID: "post-cleanup", Model: model, Input: map[string]any{"prompt": "test"}, }, sink); err == nil { t.Fatalf("expected Execute for %q to fail after cleanup, got nil", model) } } } func TestCLIStartPersistentTerminalAndExecuteWritesPrompt(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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 := 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 := &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 var deltas []string for _, e := range events { switch e.Type { case noderuntime.EventTypeStart: started = true case noderuntime.EventTypeComplete: completed = true case noderuntime.EventTypeDelta: deltas = append(deltas, e.Delta) } } if !started { t.Fatal("expected start event") } if !completed { t.Fatal("expected complete event") } combined := strings.Join(deltas, "") if !strings.Contains(combined, "reply:hello") { t.Fatalf("expected reply:hello in deltas, got %q", combined) } } // timeoutOnFirstRun ensures the context cancels during a persistent execute, // then verifies that late output from the first run does not leak into // a second run for the same profile. func TestCLIExecutePersistentTimeoutCleansSession(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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, // long idle so output doesn't timeout naturally StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() if err := c.Start(ctx); err != nil { t.Fatalf("start: %v", err) } defer func() { _ = c.Stop(ctx) }() // First run: timeout after 200ms firstCtx, firstCancel := context.WithTimeout(ctx, 200*time.Millisecond) defer firstCancel() sink1 := &fakeSink{} err := c.Execute(firstCtx, noderuntime.ExecutionSpec{ RunID: "run-1", Model: "slow-echo", Input: map[string]any{"prompt": "first"}, }, sink1) // We expect a context error (timeout or cancelled) if err == nil { t.Fatal("expected context error on first timeout execute") } // Give the background goroutine time to process any late output time.Sleep(300 * time.Millisecond) // Second run: should fail with no-session error because the session was // cleaned up after the first run's timeout. sink2 := &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") } // Verify that the second run did NOT emit any delta events (no contamination) events2 := sink2.Events() for _, e := range events2 { if e.Type == noderuntime.EventTypeDelta { t.Fatalf("unexpected delta event in second run (session contamination): %q", e.Delta) } } } // TestCLIExecutePersistentConsecutiveExecutes verifies that the same persistent // profile can be executed multiple times in a row — the second Execute must // complete (emit a Complete event) without being blocked by the first run's // session mutex. func TestCLIExecutePersistentConsecutiveExecutes(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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, // short idle so both runs complete quickly StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() if err := c.Start(ctx); err != nil { t.Fatalf("start: %v", err) } defer func() { _ = c.Stop(ctx) }() // First execute sink1 := &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() var lastType1 noderuntime.EventType for _, e := range events1 { lastType1 = e.Type } if lastType1 != noderuntime.EventTypeComplete { t.Fatalf("first run: expected last event to be complete, got %q", lastType1) } // Second execute on the same profile — must also complete without blocking. sink2 := &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() var lastType2 noderuntime.EventType for _, e := range events2 { lastType2 = e.Type } if lastType2 != noderuntime.EventTypeComplete { t.Fatalf("second run: expected last event to be complete, got %q", lastType2) } // Verify second run did NOT receive deltas from the first run. 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) } } } // TestCLIStartDeterministicOrder verifies that Start iterates profiles in // a deterministic (sorted) order by checking that cleanup after a failure stops // the earlier profiles regardless of map iteration order. func TestCLIStartDeterministicOrder(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("Unix shell required") } // "Z-alive" sorts before "Z-fail", so if deterministic order works, // Z-alive should be cleaned up when Z-fail fails. cfg := config.CLIConf{ Enabled: true, Profiles: map[string]config.CLIProfileConf{ "Z-alive": { Command: "sh", Args: []string{"-c", `while IFS= read -r line; do printf "reply:%s\n" "$line"; done`}, Persistent: true, Terminal: false, StartupIdleTimeoutMS: 50, ResponseIdleTimeoutMS: 200, }, "Z-fail": { Command: "sh", Args: []string{"-c", "exit 2"}, Persistent: true, Terminal: true, StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() if err := c.Start(ctx); err == nil { t.Fatal("expected Start to fail when one persistent profile exits immediately") } // Stop must not panic after partial start failure. _ = c.Stop(ctx) // Second Stop must also not panic (idempotency). _ = c.Stop(ctx) // All sessions must be cleared — both profiles should return no-session error. for _, model := range []string{"Z-alive", "Z-fail"} { sink := &fakeSink{} if err := c.Execute(ctx, noderuntime.ExecutionSpec{ RunID: "post-cleanup", Model: model, Input: map[string]any{"prompt": "test"}, }, sink); err == nil { t.Fatalf("expected Execute for %q to fail after cleanup, got nil", model) } } } // TestCLIConcurrentExecuteAndStop verifies that concurrent Execute and Stop // calls do not deadlock or panic, and the session map is safely protected. // Executes are launched concurrently but run sequentially due to session mutex; // the test uses short context timeouts so it completes without hanging. func TestCLIConcurrentExecuteAndStop(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") } 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.05; printf "reply:%s\n" "$line"; done`}, Persistent: true, Terminal: true, ResponseIdleTimeoutMS: 50, // short idle so completes quickly StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() if err := c.Start(ctx); err != nil { t.Fatalf("start: %v", err) } var wg sync.WaitGroup errCh := make(chan error, 10) // Goroutine 1: multiple sequential Execute calls (launched sequentially to // avoid session mutex deadlock from concurrent calls on same profile). wg.Add(1) go func() { defer wg.Done() for i := 0; i < 3; i++ { sink := &fakeSink{} execCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond) err := c.Execute(execCtx, noderuntime.ExecutionSpec{ RunID: fmt.Sprintf("run-conc-%d", i), Model: "slow-echo", Input: map[string]any{"prompt": fmt.Sprintf("conc-%d", i)}, }, sink) cancel() if err != nil { errCh <- fmt.Errorf("concurrent execute %d: %w", i, err) } } }() // Goroutine 2: Stop while executes are running wg.Add(1) go func() { defer wg.Done() time.Sleep(200 * time.Millisecond) _ = c.Stop(ctx) }() // Goroutine 3: more Execute after Stop (should get no-session error). wg.Add(1) go func() { defer wg.Done() time.Sleep(300 * time.Millisecond) for i := 0; i < 2; i++ { sink := &fakeSink{} execCtx, cancel := context.WithTimeout(ctx, 200*time.Millisecond) err := c.Execute(execCtx, noderuntime.ExecutionSpec{ RunID: fmt.Sprintf("run-post-stop-%d", i), Model: "slow-echo", Input: map[string]any{"prompt": fmt.Sprintf("post-stop-%d", i)}, }, sink) cancel() if err != nil { errCh <- fmt.Errorf("post-stop execute %d: %w", i, err) } } }() done := make(chan struct{}) go func() { wg.Wait() close(done) }() select { case <-done: close(errCh) var errs []error for err := range errCh { errs = append(errs, err) } // It's expected that some executes fail due to context timeout or no-session. // The key is that no goroutine deadlocks or panics. _ = errs case <-time.After(10 * time.Second): t.Fatal("TestCLIConcurrentExecuteAndStop timed out — possible deadlock") } } // requireEventually polls fn until it returns true or the deadline exceeds. // It helps flaky timing-dependent assertions such as process cleanup markers. func requireEventually(t *testing.T, timeout, interval time.Duration, fn func() bool) { t.Helper() timer := time.NewTimer(timeout) defer timer.Stop() ticker := time.NewTicker(interval) defer ticker.Stop() for { if fn() { return } select { case <-timer.C: t.Fatal("condition not satisfied within timeout") case <-ticker.C: } } } // readMarker returns the content of a marker file or an empty string if missing. func readMarker(path string) string { data, err := os.ReadFile(path) if err != nil { return "" } return strings.TrimSpace(string(data)) } // TestCLIStartPartialRollbackWithMarkers verifies that when a persistent profile // fails during startup, the previously started profiles are actually cleaned up // by using START_MARKER and EXIT_MARKER in their output. // // Note: SIGKILL (used by Stop/Kill) cannot be caught by trap handlers, so we send // SIGTERM first via a separate signal to trigger the EXIT trap. The test sends a // prompt to profile-a (which wakes it up) and then relies on stopAllSessions -> Kill() // which sends SIGKILL. Since SIGKILL is uncatchable, we instead verify START_MARKER // to prove the session was actually started and subsequently removed. func TestCLIStartPartialRollbackWithMarkers(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("Unix shell required") } markerDir := t.TempDir() startMarker := "START_MARKER_A" startMarkerPath := filepath.Join(markerDir, "marker_a") // First profile writes START_MARKER on start. // Uses a simple approach: write marker immediately on startup. profileACmd := fmt.Sprintf(`mkdir -p %s; echo %s > %s/marker_a; while IFS= read -r line; do printf "reply:%%s\n" "$line"; done`, markerDir, startMarker, markerDir) cfg := config.CLIConf{ Enabled: true, Profiles: map[string]config.CLIProfileConf{ "profile-a": { Command: "sh", Args: []string{"-c", profileACmd}, Persistent: true, Terminal: true, // PTY so the process starts properly StartupIdleTimeoutMS: 50, ResponseIdleTimeoutMS: 200, }, "profile-b": { Command: "sh", Args: []string{"-c", "exit 2"}, Persistent: true, Terminal: true, StartupIdleTimeoutMS: 50, }, }, } c := New(cfg, zap.NewNop()) ctx := context.Background() // Start should fail because profile-b exits immediately. if err := c.Start(ctx); err == nil { t.Fatal("expected Start to fail when profile-b exits immediately") } // Stop must not panic after partial start failure. _ = c.Stop(ctx) // Give the cleanup goroutine time to finish. time.Sleep(300 * time.Millisecond) // Assertion 1: START_MARKER must exist (profile-a started successfully). requireEventually(t, 2*time.Second, 50*time.Millisecond, func() bool { content := readMarker(startMarkerPath) return content == startMarker }) // Assertion 2: the session was removed — Execute should fail with no-session. sink := &fakeSink{} if err := c.Execute(ctx, noderuntime.ExecutionSpec{ RunID: "post-rollback", Model: "profile-a", Input: map[string]any{"prompt": "test"}, }, sink); err == nil { t.Fatal("expected Execute for profile-a to fail with no-session after rollback") } }