toki
4bece3fb2d
- [2-1] executePersistent: add defer sess.mu.Unlock() so all return paths release the session mutex. Remove explicit unlock calls. - [2-2] Protect c.sessions map access in Execute with c.mu. Replace c.Stop() in timeout/cancel path with per-session close/kill/delete so only the affected profile session is cleaned up. - [2-3] Add TestCLIExecutePersistentConsecutiveExecutes to verify the same persistent profile can be executed consecutively without blocking. Apply gofmt to all adapter files.
523 lines
14 KiB
Go
523 lines
14 KiB
Go
package cli_test
|
|
|
|
import (
|
|
"context"
|
|
"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 order by checking that cleanup after a failure stops
|
|
// the earlier profiles regardless of map iteration order.
|
|
func TestCLIStartPartialRollbackChecksMarker(t *testing.T) {
|
|
if runtime.GOOS == "windows" {
|
|
t.Skip("Unix shell required")
|
|
}
|
|
|
|
// Use a profile that logs a start marker and waits for input, so it stays alive.
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"alive-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)
|
|
|
|
// All sessions must be cleared — both profiles should return no-session error.
|
|
for _, model := range []string{"alive-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)
|
|
}
|
|
}
|
|
}
|