// Package cli provides an Adapter that runs external CLI tools as inference // backends. Profiles (claude, gemini, codex, opencode) are configured in // configs/node.yaml and select the command + args to execute. package cli import ( "bufio" "context" "fmt" "io" "os/exec" "strings" "sync" "time" "github.com/creack/pty" "go.uber.org/zap" "iop/apps/node/internal/runtime" "iop/packages/config" ) 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, } } func (c *CLI) Name() string { return Name } func (c *CLI) Capabilities(_ context.Context) (runtime.Capabilities, error) { models := make([]string, 0, len(c.profiles)) for name := range c.profiles { models = append(models, name) } // Always advertise known profiles even if not in config yet. for _, p := range knownProfiles { if _, ok := c.profiles[p]; !ok { models = append(models, p) } } return runtime.Capabilities{ AdapterName: Name, Models: models, MaxConcurrency: 4, }, 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...) if len(profile.Env) > 0 { cmd.Env = append(cmd.Environ(), profile.Env...) } stdout, err := cmd.StdoutPipe() if err != nil { return fmt.Errorf("cli adapter: stdout pipe: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { return fmt.Errorf("cli adapter: stderr pipe: %w", err) } if err := cmd.Start(); err != nil { return fmt.Errorf("cli adapter: start %q: %w", profile.Command, err) } _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeStart, Timestamp: time.Now(), }) scanner := bufio.NewScanner(stdout) outputTokens := 0 for scanner.Scan() { line := scanner.Text() outputTokens += len(strings.Fields(line)) _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeDelta, Delta: line + "\n", Timestamp: time.Now(), }) } var errBuf strings.Builder errScanner := bufio.NewScanner(stderr) for errScanner.Scan() { errBuf.WriteString(errScanner.Text()) errBuf.WriteByte('\n') } if err := cmd.Wait(); err != nil { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("command failed: %v — %s", err, errBuf.String()), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: command exited with error: %w", err) } 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(), }) } 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 "" } if v, ok := input["prompt"]; ok { if s, ok := v.(string); ok { return s } } return fmt.Sprintf("%v", input) }