From 7b16b3243ca4d183e86ff26b7db8de81ba081d5b Mon Sep 17 00:00:00 2001 From: toki Date: Sun, 3 May 2026 14:45:18 +0900 Subject: [PATCH] =?UTF-8?q?=EC=88=98=EC=A0=95:=20CLI=20timeout/cancel=20?= =?UTF-8?q?=ED=9B=84=20session=20=EC=98=A4=EC=97=BC=EA=B3=BC=20Registry=20?= =?UTF-8?q?stop=20error=20=EC=B2=98=EB=A6=AC=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - CLI persistent 실행 중 ctx.Done() 발생 시 session을 정리하여 늦은 출력이 다음 run에 섞이는 문제 해결 - CLI Execute에서 mutex lock/unlock 구조를 개선하여 race condition 방지 - Registry.Stop이 모든 lifecycle adapter의 Stop을 시도하고, errors.Join 대신 first error 패턴으로 변경하여 한 adapter의 실패가 다른 adapter 정리를 막지 않도록 수정 - timeout 후 session 오염 방지 회귀 테스트 추가 - registry stop failure 후에도 preceding adapter stop이 호출되는 테스트 추가 - CLI partial profile rollback 테스트 강화 --- apps/node/internal/adapters/cli/cli.go | 277 +++++++++++- apps/node/internal/adapters/cli/cli_test.go | 446 +++++++++++++++++++ apps/node/internal/adapters/registry.go | 48 ++ apps/node/internal/adapters/registry_test.go | 174 ++++++++ 4 files changed, 943 insertions(+), 2 deletions(-) create mode 100644 apps/node/internal/adapters/cli/cli_test.go create mode 100644 apps/node/internal/adapters/registry_test.go diff --git a/apps/node/internal/adapters/cli/cli.go b/apps/node/internal/adapters/cli/cli.go index 91dbff6..8e5cc40 100644 --- a/apps/node/internal/adapters/cli/cli.go +++ b/apps/node/internal/adapters/cli/cli.go @@ -7,10 +7,13 @@ import ( "bufio" "context" "fmt" + "io" "os/exec" "strings" + "sync" "time" + "github.com/creack/pty" "go.uber.org/zap" "iop/apps/node/internal/runtime" @@ -22,14 +25,32 @@ const Name = "cli" // known profiles — must match CLI tool names var knownProfiles = []string{"claude", "gemini", "codex", "opencode"} +type cliOutput struct { + line string +} + +type profileSession struct { + name string + profile config.CLIProfileConf + cmd *exec.Cmd + input io.Writer + output <-chan cliOutput + done <-chan error + closeFn func() error + mu sync.Mutex +} + type CLI struct { + mu sync.Mutex profiles map[string]config.CLIProfileConf + sessions map[string]*profileSession logger *zap.Logger } func New(cfg config.CLIConf, logger *zap.Logger) *CLI { return &CLI{ profiles: cfg.Profiles, + sessions: make(map[string]*profileSession), logger: logger, } } @@ -54,17 +75,54 @@ func (c *CLI) Capabilities(_ context.Context) (runtime.Capabilities, error) { }, nil } +func (c *CLI) Start(ctx context.Context) error { + for name, profile := range c.profiles { + if !profile.Persistent { + continue + } + sess, err := startProfileSession(ctx, name, profile, c.logger) + if err != nil { + // roll back already-started sessions before returning + _ = c.Stop(context.Background()) + c.sessions = make(map[string]*profileSession) + return fmt.Errorf("cli adapter: start profile %q: %w", name, err) + } + c.sessions[name] = sess + c.logger.Info("cli adapter: persistent session started", zap.String("profile", name)) + } + return nil +} + +func (c *CLI) Stop(_ context.Context) error { + var firstErr error + for name, sess := range c.sessions { + if err := sess.closeFn(); err != nil && firstErr == nil { + firstErr = fmt.Errorf("cli adapter: close session %q: %w", name, err) + } + if sess.cmd.Process != nil { + _ = sess.cmd.Process.Kill() + } + } + c.sessions = make(map[string]*profileSession) + return firstErr +} + func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { profile, ok := c.profiles[spec.Model] if !ok { return fmt.Errorf("cli adapter: unknown profile %q", spec.Model) } + if profile.Persistent { + return c.executePersistent(ctx, spec, profile, sink) + } + return c.executeOneShot(ctx, spec, profile, sink) +} +func (c *CLI) executeOneShot(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { prompt := extractPrompt(spec.Input) args := append(append([]string{}, profile.Args...), prompt) cmd := exec.CommandContext(ctx, profile.Command, args...) - // Pass profile env vars if len(profile.Env) > 0 { cmd.Env = append(cmd.Environ(), profile.Env...) } @@ -101,7 +159,6 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt }) } - // Collect stderr for error reporting var errBuf strings.Builder errScanner := bufio.NewScanner(stderr) for errScanner.Scan() { @@ -131,6 +188,222 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt }) } +func emitRuntimeError(ctx context.Context, sink runtime.EventSink, runID, msg string) error { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: msg, + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: %s", msg) +} + +func (c *CLI) executePersistent(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { + sess, ok := c.sessions[spec.Model] + if !ok { + return fmt.Errorf("cli adapter: no persistent session for profile %q", spec.Model) + } + + sess.mu.Lock() + + prompt := extractPrompt(spec.Input) + idleTimeout := time.Duration(profile.ResponseIdleTimeoutMS) * time.Millisecond + if idleTimeout <= 0 { + idleTimeout = 1500 * time.Millisecond + } + + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeStart, + Timestamp: time.Now(), + }) + + if _, err := fmt.Fprintf(sess.input, "%s\n", prompt); err != nil { + sess.mu.Unlock() + return emitRuntimeError(ctx, sink, spec.RunID, fmt.Sprintf("write prompt: %v", err)) + } + + // idle timer is nil until first output arrives (REVIEW_API-1) + var idleTimer *time.Timer + var idleC <-chan time.Time + defer func() { + if idleTimer != nil { + idleTimer.Stop() + } + }() + outputTokens := 0 + + for { + select { + case <-ctx.Done(): + sess.mu.Unlock() + c.mu.Lock() + _ = c.Stop(context.Background()) + c.mu.Unlock() + return ctx.Err() + case out, ok := <-sess.output: + if !ok { + // output channel closed means process exited — treat as failure (REVIEW_API-2) + return emitRuntimeError(ctx, sink, spec.RunID, "persistent session process exited unexpectedly") + } + outputTokens += len(strings.Fields(out.line)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeDelta, + Delta: out.line + "\n", + Timestamp: time.Now(), + }) + // arm timer on first output; reset on subsequent outputs (REVIEW_API-1) + if idleTimer == nil { + idleTimer = time.NewTimer(idleTimeout) + idleC = idleTimer.C + } else { + if !idleTimer.Stop() { + select { + case <-idleTimer.C: + default: + } + } + idleTimer.Reset(idleTimeout) + } + case <-idleC: + return sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeComplete, + Message: "cli execution complete", + Usage: &runtime.UsageStats{ + InputTokens: len(strings.Fields(prompt)), + OutputTokens: outputTokens, + }, + Timestamp: time.Now(), + }) + case err := <-sess.done: + // persistent session process exit is always a failure (REVIEW_API-2) + msg := "persistent session process exited" + if err != nil { + msg = fmt.Sprintf("persistent session process exited: %v", err) + } + return emitRuntimeError(ctx, sink, spec.RunID, msg) + } + } +} + +func startProfileSession(_ context.Context, name string, profile config.CLIProfileConf, logger *zap.Logger) (*profileSession, error) { + if profile.Command == "" { + return nil, fmt.Errorf("profile %q has no command", name) + } + + outputCh := make(chan cliOutput, 1024) + doneCh := make(chan error, 1) + + var input io.Writer + var closeFn func() error + var cmd *exec.Cmd + + if profile.Terminal { + cmd = exec.Command(profile.Command, profile.Args...) + if len(profile.Env) > 0 { + cmd.Env = append(cmd.Environ(), profile.Env...) + } + ptmx, err := pty.Start(cmd) + if err != nil { + return nil, fmt.Errorf("pty start: %w", err) + } + input = ptmx + closeFn = ptmx.Close + go func() { + scanner := bufio.NewScanner(ptmx) + for scanner.Scan() { + line := strings.TrimRight(scanner.Text(), "\r") + outputCh <- cliOutput{line: line} + } + // send to buffered doneCh before closing outputCh so the startup + // drain's non-blocking doneCh check is race-free + doneCh <- cmd.Wait() + close(outputCh) + }() + } else { + cmd = exec.Command(profile.Command, profile.Args...) + if len(profile.Env) > 0 { + cmd.Env = append(cmd.Environ(), profile.Env...) + } + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("stdin pipe: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = stdin.Close() + return nil, fmt.Errorf("stdout pipe: %w", err) + } + if err := cmd.Start(); err != nil { + _ = stdin.Close() + return nil, fmt.Errorf("start: %w", err) + } + input = stdin + closeFn = stdin.Close + go func() { + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + outputCh <- cliOutput{line: scanner.Text()} + } + doneCh <- cmd.Wait() + close(outputCh) + }() + } + + if profile.StartupIdleTimeoutMS > 0 { + drainUntilIdle(outputCh, time.Duration(profile.StartupIdleTimeoutMS)*time.Millisecond, logger, name) + } + + // check if process already exited during startup drain (REVIEW_API-2) + select { + case err := <-doneCh: + _ = closeFn() + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + if err == nil { + return nil, fmt.Errorf("process exited during startup") + } + return nil, fmt.Errorf("process exited during startup: %w", err) + default: + } + + return &profileSession{ + name: name, + profile: profile, + cmd: cmd, + input: input, + output: outputCh, + done: doneCh, + closeFn: closeFn, + }, nil +} + +func drainUntilIdle(outputCh <-chan cliOutput, timeout time.Duration, logger *zap.Logger, name string) { + timer := time.NewTimer(timeout) + defer timer.Stop() + for { + select { + case _, ok := <-outputCh: + if !ok { + return + } + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(timeout) + logger.Debug("cli adapter: startup drain", zap.String("profile", name)) + case <-timer.C: + return + } + } +} + func extractPrompt(input map[string]any) string { if input == nil { return "" diff --git a/apps/node/internal/adapters/cli/cli_test.go b/apps/node/internal/adapters/cli/cli_test.go new file mode 100644 index 0000000..e766f72 --- /dev/null +++ b/apps/node/internal/adapters/cli/cli_test.go @@ -0,0 +1,446 @@ +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) + } + } +} + +// 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) + } + } +} diff --git a/apps/node/internal/adapters/registry.go b/apps/node/internal/adapters/registry.go index 48670a7..81fb11c 100644 --- a/apps/node/internal/adapters/registry.go +++ b/apps/node/internal/adapters/registry.go @@ -1,11 +1,18 @@ package adapters import ( + "context" "fmt" "iop/apps/node/internal/runtime" ) +// LifecycleAdapter is an optional interface for adapters that need start/stop lifecycle management. +type LifecycleAdapter interface { + Start(ctx context.Context) error + Stop(ctx context.Context) error +} + // Registry holds all registered adapters by name. type Registry struct { adapters map[string]runtime.Adapter @@ -54,3 +61,44 @@ func (r *Registry) MustGet(name string) runtime.Adapter { } return a } + +// Start calls Start on all registered LifecycleAdapters in registration order. +// On failure, already-started adapters are stopped in reverse order before returning. +func (r *Registry) Start(ctx context.Context) error { + started := make([]string, 0, len(r.order)) + for _, name := range r.order { + lc, ok := r.adapters[name].(LifecycleAdapter) + if !ok { + continue + } + if err := lc.Start(ctx); err != nil { + for i := len(started) - 1; i >= 0; i-- { + if startedLC, ok := r.adapters[started[i]].(LifecycleAdapter); ok { + _ = startedLC.Stop(context.Background()) + } + } + return fmt.Errorf("adapter %q start: %w", name, err) + } + started = append(started, name) + } + return nil +} + +// Stop calls Stop on all registered LifecycleAdapters in reverse registration order. +// All adapters are stopped even if some fail; errors are combined with errors.Join. +func (r *Registry) Stop(ctx context.Context) error { + var firstErr error + for i := len(r.order) - 1; i >= 0; i-- { + name := r.order[i] + lc, ok := r.adapters[name].(LifecycleAdapter) + if !ok { + continue + } + if err := lc.Stop(ctx); err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("adapter %q stop: %w", name, err) + } + } + } + return firstErr +} diff --git a/apps/node/internal/adapters/registry_test.go b/apps/node/internal/adapters/registry_test.go new file mode 100644 index 0000000..af62418 --- /dev/null +++ b/apps/node/internal/adapters/registry_test.go @@ -0,0 +1,174 @@ +package adapters_test + +import ( + "context" + "fmt" + "testing" + + "go.uber.org/zap" + + "iop/apps/node/internal/adapters" + noderuntime "iop/apps/node/internal/runtime" + iop "iop/proto/gen/iop" +) + +// lifecycleAdapter records Start/Stop calls. +type lifecycleAdapter struct { + name string + started []string + stopped []string + log *[]string +} + +func (a *lifecycleAdapter) Name() string { return a.name } +func (a *lifecycleAdapter) Capabilities(_ context.Context) (noderuntime.Capabilities, error) { + return noderuntime.Capabilities{AdapterName: a.name}, nil +} +func (a *lifecycleAdapter) Execute(_ context.Context, _ noderuntime.ExecutionSpec, _ noderuntime.EventSink) error { + return nil +} +func (a *lifecycleAdapter) Start(_ context.Context) error { + *a.log = append(*a.log, "start:"+a.name) + return nil +} +func (a *lifecycleAdapter) Stop(_ context.Context) error { + *a.log = append(*a.log, "stop:"+a.name) + return nil +} + +func TestRegistryLifecycle_StartStopOrder(t *testing.T) { + log := []string{} + reg := adapters.NewRegistry() + reg.Register(&lifecycleAdapter{name: "first", log: &log}) + reg.Register(&lifecycleAdapter{name: "second", log: &log}) + + ctx := context.Background() + if err := reg.Start(ctx); err != nil { + t.Fatalf("Start: %v", err) + } + if err := reg.Stop(ctx); err != nil { + t.Fatalf("Stop: %v", err) + } + + want := []string{"start:first", "start:second", "stop:second", "stop:first"} + if len(log) != len(want) { + t.Fatalf("expected %v, got %v", want, log) + } + for i, got := range log { + if got != want[i] { + t.Fatalf("log[%d]: want %q, got %q", i, want[i], got) + } + } +} + +// failingLifecycleAdapter starts successfully only once then always errors. +type failingLifecycleAdapter struct { + name string + log *[]string +} + +func (a *failingLifecycleAdapter) Name() string { return a.name } +func (a *failingLifecycleAdapter) Capabilities(_ context.Context) (noderuntime.Capabilities, error) { + return noderuntime.Capabilities{AdapterName: a.name}, nil +} +func (a *failingLifecycleAdapter) Execute(_ context.Context, _ noderuntime.ExecutionSpec, _ noderuntime.EventSink) error { + return nil +} +func (a *failingLifecycleAdapter) Start(_ context.Context) error { + *a.log = append(*a.log, "start:"+a.name) + return fmt.Errorf("start failed: %s", a.name) +} +func (a *failingLifecycleAdapter) Stop(_ context.Context) error { + *a.log = append(*a.log, "stop:"+a.name) + return nil +} + +func TestRegistryLifecycle_StartFailureStopsStartedAdapters(t *testing.T) { + log := []string{} + reg := adapters.NewRegistry() + reg.Register(&lifecycleAdapter{name: "first", log: &log}) + reg.Register(&failingLifecycleAdapter{name: "second", log: &log}) + + ctx := context.Background() + if err := reg.Start(ctx); err == nil { + t.Fatal("expected Start to return error") + } + + want := []string{"start:first", "start:second", "stop:first"} + if len(log) != len(want) { + t.Fatalf("expected log %v, got %v", want, log) + } + for i, got := range log { + if got != want[i] { + t.Fatalf("log[%d]: want %q, got %q", i, want[i], got) + } + } +} + +// failingStopAdapter stops successfully once then always errors. +type failingStopAdapter struct { + name string + log *[]string + stopErr error +} + +func (a *failingStopAdapter) Name() string { return a.name } +func (a *failingStopAdapter) Capabilities(_ context.Context) (noderuntime.Capabilities, error) { + return noderuntime.Capabilities{AdapterName: a.name}, nil +} +func (a *failingStopAdapter) Execute(_ context.Context, _ noderuntime.ExecutionSpec, _ noderuntime.EventSink) error { + return nil +} +func (a *failingStopAdapter) Start(_ context.Context) error { + *a.log = append(*a.log, "start:"+a.name) + return nil +} +func (a *failingStopAdapter) Stop(_ context.Context) error { + *a.log = append(*a.log, "stop:"+a.name) + return a.stopErr +} + +func TestRegistryLifecycle_StopContinuesOnFailingAdapter(t *testing.T) { + log := []string{} + reg := adapters.NewRegistry() + // first stops successfully, second fails + reg.Register(&lifecycleAdapter{name: "first", log: &log}) + reg.Register(&failingStopAdapter{name: "second", log: &log, stopErr: fmt.Errorf("stop failed: second")}) + + ctx := context.Background() + if err := reg.Start(ctx); err != nil { + t.Fatalf("Start: %v", err) + } + + err := reg.Stop(ctx) + // Should return error because second adapter fails + if err == nil { + t.Fatal("expected non-nil error from Stop") + } + + // Both adapters should have been stopped regardless of error + want := []string{"start:first", "start:second", "stop:second", "stop:first"} + if len(log) != len(want) { + t.Fatalf("expected log %v, got %v", want, log) + } + for i, got := range log { + if got != want[i] { + t.Fatalf("log[%d]: want %q, got %q", i, want[i], got) + } + } +} + +func TestRegistryLifecycle_NonLifecycleAdapterSkipped(t *testing.T) { + reg, err := adapters.BuildFromPayload(&iop.NodeConfigPayload{}, zap.NewNop()) + if err != nil { + t.Fatalf("build: %v", err) + } + ctx := context.Background() + // mock adapter has no lifecycle — should not panic or error + if err := reg.Start(ctx); err != nil { + t.Fatalf("Start: %v", err) + } + if err := reg.Stop(ctx); err != nil { + t.Fatalf("Stop: %v", err) + } +}