iop/apps/node/internal/adapters/cli/cli.go

// 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)
}