From 0988942b074b0db44cccc6b6c83fcc5ce29d10e7 Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 3 May 2026 18:19:00 +0900 Subject: [PATCH] refactor: reorganize CLI adapter into subpackages - Move cli_test.go to internal/, lifecycle/, onshot/, persistent/ - Restructure CLI adapter modules into organized subdirectories --- apps/node/internal/adapters/cli/cli_test.go | 773 ------------------ .../cli/internal/testutil/testutil.go | 85 ++ .../adapters/cli/lifecycle/cli_test.go | 261 ++++++ .../internal/adapters/cli/oneshot/cli_test.go | 94 +++ .../adapters/cli/persistent/execute_test.go | 280 +++++++ 5 files changed, 720 insertions(+), 773 deletions(-) delete mode 100644 apps/node/internal/adapters/cli/cli_test.go create mode 100644 apps/node/internal/adapters/cli/internal/testutil/testutil.go create mode 100644 apps/node/internal/adapters/cli/lifecycle/cli_test.go create mode 100644 apps/node/internal/adapters/cli/oneshot/cli_test.go create mode 100644 apps/node/internal/adapters/cli/persistent/execute_test.go diff --git a/apps/node/internal/adapters/cli/cli_test.go b/apps/node/internal/adapters/cli/cli_test.go deleted file mode 100644 index c3eccd7..0000000 --- a/apps/node/internal/adapters/cli/cli_test.go +++ /dev/null @@ -1,773 +0,0 @@ -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 TestCLIExecuteOneShotDrainsStderrConcurrently(t *testing.T) { - if runtime.GOOS == "windows" { - t.Skip("sh not available on Windows") - } - cfg := config.CLIConf{ - Enabled: true, - Profiles: map[string]config.CLIProfileConf{ - "noisy": { - Command: "sh", - Args: []string{ - "-c", - `i=0; while [ "$i" -lt 20000 ]; do echo "warning line $i" >&2; i=$((i+1)); done; printf "reply:%s\n" "$1"`, - "sh", - }, - }, - }, - } - c := New(cfg, zap.NewNop()) - sink := &fakeSink{} - - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() - err := c.Execute(ctx, noderuntime.ExecutionSpec{ - RunID: "run-noisy", - Model: "noisy", - Input: map[string]any{"prompt": "hello"}, - }, sink) - if err != nil { - t.Fatalf("execute: %v", err) - } - - var deltas []string - for _, e := range sink.Events() { - if e.Type == noderuntime.EventTypeDelta { - deltas = append(deltas, e.Delta) - } - } - if combined := strings.Join(deltas, ""); !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") - } -} diff --git a/apps/node/internal/adapters/cli/internal/testutil/testutil.go b/apps/node/internal/adapters/cli/internal/testutil/testutil.go new file mode 100644 index 0000000..2ba6b1d --- /dev/null +++ b/apps/node/internal/adapters/cli/internal/testutil/testutil.go @@ -0,0 +1,85 @@ +package testutil + +import ( + "context" + "os" + "runtime" + "strings" + "sync" + "testing" + "time" + + noderuntime "iop/apps/node/internal/runtime" +) + +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 RequireUnixShell(t *testing.T) { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("Unix shell required") + } +} + +func RequirePTYSupport(t *testing.T) { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("PTY not supported on Windows") + } +} + +func CollectDeltas(events []noderuntime.RuntimeEvent) string { + deltas := make([]string, 0, len(events)) + for _, e := range events { + if e.Type == noderuntime.EventTypeDelta { + deltas = append(deltas, e.Delta) + } + } + return strings.Join(deltas, "") +} + +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: + } + } +} + +func ReadMarker(path string) string { + data, err := os.ReadFile(path) + if err != nil { + return "" + } + return strings.TrimSpace(string(data)) +} diff --git a/apps/node/internal/adapters/cli/lifecycle/cli_test.go b/apps/node/internal/adapters/cli/lifecycle/cli_test.go new file mode 100644 index 0000000..51b011f --- /dev/null +++ b/apps/node/internal/adapters/cli/lifecycle/cli_test.go @@ -0,0 +1,261 @@ +package lifecycle_test + +import ( + "context" + "fmt" + "path/filepath" + "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 TestCLIStartPersistentReturnsErrorWhenProcessExitsDuringStartup(t *testing.T) { + testutil.RequirePTYSupport(t) + + 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 := clipkg.New(cfg, zap.NewNop()) + + if err := c.Start(context.Background()); err == nil { + t.Fatal("expected error when persistent process exits during startup") + } +} + +func TestCLIStartCleansUpStartedProfilesWhenLaterProfileFails(t *testing.T) { + testutil.RequireUnixShell(t) + + 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 := clipkg.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") + } + + _ = c.Stop(ctx) + _ = c.Stop(ctx) + + for _, model := range []string{"ok-profile", "fail-profile"} { + sink := &testutil.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 TestCLIStartDeterministicOrder(t *testing.T) { + testutil.RequireUnixShell(t) + + 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 := clipkg.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") + } + + _ = c.Stop(ctx) + _ = c.Stop(ctx) + + for _, model := range []string{"Z-alive", "Z-fail"} { + sink := &testutil.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 TestCLIConcurrentExecuteAndStop(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.05; 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) + } + + errCh := make(chan error, 10) + done := make(chan struct{}) + + go func() { + for i := 0; i < 3; i++ { + sink := &testutil.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) + } + } + done <- struct{}{} + }() + + go func() { + time.Sleep(200 * time.Millisecond) + _ = c.Stop(ctx) + done <- struct{}{} + }() + + go func() { + time.Sleep(300 * time.Millisecond) + for i := 0; i < 2; i++ { + sink := &testutil.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 <- struct{}{} + }() + + timeout := time.After(10 * time.Second) + for completed := 0; completed < 3; completed++ { + select { + case <-done: + case <-timeout: + t.Fatal("TestCLIConcurrentExecuteAndStop timed out; possible deadlock") + } + } + close(errCh) + for range errCh { + } +} + +func TestCLIStartPartialRollbackWithMarkers(t *testing.T) { + testutil.RequireUnixShell(t) + + markerDir := t.TempDir() + startMarker := "START_MARKER_A" + startMarkerPath := filepath.Join(markerDir, "marker_a") + 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, + StartupIdleTimeoutMS: 50, + ResponseIdleTimeoutMS: 200, + }, + "profile-b": { + Command: "sh", + Args: []string{"-c", "exit 2"}, + Persistent: true, + Terminal: true, + StartupIdleTimeoutMS: 50, + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + ctx := context.Background() + + if err := c.Start(ctx); err == nil { + t.Fatal("expected Start to fail when profile-b exits immediately") + } + + _ = c.Stop(ctx) + time.Sleep(300 * time.Millisecond) + + testutil.RequireEventually(t, 2*time.Second, 50*time.Millisecond, func() bool { + return testutil.ReadMarker(startMarkerPath) == startMarker + }) + + sink := &testutil.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") + } +} diff --git a/apps/node/internal/adapters/cli/oneshot/cli_test.go b/apps/node/internal/adapters/cli/oneshot/cli_test.go new file mode 100644 index 0000000..c7e5652 --- /dev/null +++ b/apps/node/internal/adapters/cli/oneshot/cli_test.go @@ -0,0 +1,94 @@ +package oneshot_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 TestCLIExecuteOneShotPassesPromptAsArg(t *testing.T) { + testutil.RequireUnixShell(t) + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "echo": { + Command: "sh", + Args: []string{"-c", `printf "reply:%s\n" "$1"`, "sh"}, + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.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 + for _, e := range events { + types = append(types, string(e.Type)) + } + 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) + } + if combined := testutil.CollectDeltas(events); !strings.Contains(combined, "reply:hello") { + t.Fatalf("expected reply:hello in deltas, got %q", combined) + } +} + +func TestCLIExecuteOneShotDrainsStderrConcurrently(t *testing.T) { + testutil.RequireUnixShell(t) + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "noisy": { + Command: "sh", + Args: []string{ + "-c", + `i=0; while [ "$i" -lt 20000 ]; do echo "warning line $i" >&2; i=$((i+1)); done; printf "reply:%s\n" "$1"`, + "sh", + }, + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-noisy", + Model: "noisy", + Input: map[string]any{"prompt": "hello"}, + }, sink) + if err != nil { + t.Fatalf("execute: %v", err) + } + + if combined := testutil.CollectDeltas(sink.Events()); !strings.Contains(combined, "reply:hello") { + t.Fatalf("expected reply:hello in deltas, got %q", combined) + } +} diff --git a/apps/node/internal/adapters/cli/persistent/execute_test.go b/apps/node/internal/adapters/cli/persistent/execute_test.go new file mode 100644 index 0000000..e74df84 --- /dev/null +++ b/apps/node/internal/adapters/cli/persistent/execute_test.go @@ -0,0 +1,280 @@ +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) + } + } +}