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

// Package cli provides an Adapter that runs external CLI tools as an execution
// adapter. Profiles (claude, gemini, codex, opencode, cline) are configured in
// configs/node.yaml and select the command + args to execute. Interactive CLIs
// should be configured with their non-interactive/headless flags for use in the
// request/response node pipeline.
package cli

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"sort"
	"sync"

	"go.uber.org/zap"

	"iop/apps/node/internal/adapters/cli/status"
	"iop/apps/node/internal/runtime"
	"iop/packages/config"
)

const Name = "cli"

type cliOutput struct {
	line string
}

// sessionKey uniquely identifies a logical worker session.
type sessionKey struct {
	agent     string
	sessionID string
}

type profileSession struct {
	key     sessionKey
	name    string
	profile config.CLIProfileConf
	cmd     *exec.Cmd
	input   io.Writer
	output  <-chan cliOutput
	done    <-chan error
	closeFn func() error
	mu      sync.Mutex
}

type codexExecSession struct {
	key        sessionKey
	externalID string
	mu         sync.Mutex
}

type CLI struct {
	mu            sync.Mutex
	profiles      map[string]config.CLIProfileConf
	sessions      map[sessionKey]*profileSession
	codexSessions map[sessionKey]*codexExecSession
	logger        *zap.Logger

	// StatusChecker overrides status.CheckUsage for testing.
	StatusChecker func(ctx context.Context, agent string, profile config.CLIProfileConf) (*status.UsageStatus, error)
}

func New(cfg config.CLIConf, logger *zap.Logger) *CLI {
	return &CLI{
		profiles:      cfg.Profiles,
		sessions:      make(map[sessionKey]*profileSession),
		codexSessions: make(map[sessionKey]*codexExecSession),
		logger:        logger,
	}
}

func (c *CLI) Name() string { return Name }

func (c *CLI) Capabilities(_ context.Context) (runtime.Capabilities, error) {
	profiles := make([]string, 0, len(c.profiles))
	for name := range c.profiles {
		profiles = append(profiles, name)
	}
	sort.Strings(profiles)
	return runtime.Capabilities{
		AdapterName:    Name,
		Targets:        profiles,
		MaxConcurrency: 4,
	}, nil
}

// Start starts the default session for each persistent profile in deterministic (sorted) order.
// On failure, already-started sessions are rolled back under c.mu.
func (c *CLI) Start(ctx context.Context) error {
	names := make([]string, 0, len(c.profiles))
	for name := range c.profiles {
		if c.profiles[name].Persistent && c.profiles[name].Mode != "codex-exec" {
			names = append(names, name)
		}
	}
	sort.Strings(names)

	for _, name := range names {
		profile := c.profiles[name]
		key := sessionKey{agent: name, sessionID: runtime.DefaultSessionID}
		sess, err := startProfileSession(ctx, key, profile, c.logger)
		if err != nil {
			c.mu.Lock()
			_ = c.stopAllSessions(context.Background())
			c.sessions = make(map[sessionKey]*profileSession)
			c.mu.Unlock()
			return fmt.Errorf("cli adapter: start agent %q: %w", name, err)
		}
		c.mu.Lock()
		c.sessions[key] = sess
		c.mu.Unlock()
		c.logger.Info("cli adapter: persistent session started", zap.String("agent", name))
	}
	return nil
}

// stopAllSessions closes all sessions and clears the map.
// Must be called while c.mu is held.
func (c *CLI) stopAllSessions(_ context.Context) error {
	var firstErr error
	for key, sess := range c.sessions {
		if err := closeProfileSession(context.Background(), sess); err != nil && firstErr == nil {
			firstErr = fmt.Errorf("cli adapter: close session %q/%q: %w", key.agent, key.sessionID, err)
		}
	}
	c.sessions = make(map[sessionKey]*profileSession)
	return firstErr
}

// Stop stops all logical sessions. Errors are combined by reporting only the first.
func (c *CLI) Stop(_ context.Context) error {
	c.mu.Lock()
	sessionsCopy := make(map[sessionKey]*profileSession, len(c.sessions))
	for key, sess := range c.sessions {
		sessionsCopy[key] = sess
	}
	c.sessions = make(map[sessionKey]*profileSession)
	c.codexSessions = make(map[sessionKey]*codexExecSession)
	c.mu.Unlock()

	var firstErr error
	for key, sess := range sessionsCopy {
		if err := closeProfileSession(context.Background(), sess); err != nil && firstErr == nil {
			firstErr = fmt.Errorf("cli adapter: close session %q/%q: %w", key.agent, key.sessionID, err)
		}
	}
	return firstErr
}

func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error {
	agent := cliAgentName(spec)
	profile, ok := c.profiles[agent]
	if !ok {
		return fmt.Errorf("cli adapter: unknown agent %q", agent)
	}
	if profile.Mode == "codex-exec" {
		return c.executeCodexExec(ctx, spec, profile, sink)
	}
	if profile.Persistent {
		return c.executePersistent(ctx, spec, profile, sink)
	}
	return c.executeOneShot(ctx, spec, profile, sink)
}

func (c *CLI) HandleCommand(ctx context.Context, req runtime.CommandRequest) (runtime.CommandResponse, error) {
	if req.Type != runtime.CommandTypeUsageStatus {
		return runtime.CommandResponse{}, fmt.Errorf("cli adapter: unsupported command %q", req.Type)
	}
	profile, ok := c.profiles[req.Target]
	if !ok {
		return runtime.CommandResponse{}, fmt.Errorf("cli adapter: unknown agent %q", req.Target)
	}

	checkFn := c.StatusChecker
	if checkFn == nil {
		checkFn = status.CheckUsage
	}
	st, err := checkFn(ctx, req.Target, profile)
	if err != nil {
		return runtime.CommandResponse{}, err
	}
	return runtime.CommandResponse{
		RequestID:   req.RequestID,
		Type:        req.Type,
		Adapter:     req.Adapter,
		Target:      req.Target,
		SessionID:   req.SessionID,
		UsageStatus: st.ToRuntime(),
	}, nil
}

// TerminateSession implements runtime.SessionTerminator.
func (c *CLI) TerminateSession(_ context.Context, agent, sessionID string) error {
	key := sessionKey{agent: agent, sessionID: normalizeSessionID(sessionID)}
	if profile, ok := c.profiles[agent]; ok && profile.Mode == "codex-exec" {
		c.mu.Lock()
		_, ok := c.codexSessions[key]
		if ok {
			delete(c.codexSessions, key)
		}
		c.mu.Unlock()
		if !ok {
			return fmt.Errorf("cli adapter: no session %q for agent %q", key.sessionID, agent)
		}
		return nil
	}

	c.mu.Lock()
	sess, ok := c.sessions[key]
	if ok {
		delete(c.sessions, key)
	}
	c.mu.Unlock()
	if !ok {
		return fmt.Errorf("cli adapter: no session %q for agent %q", key.sessionID, agent)
	}
	return closeProfileSession(context.Background(), sess)
}

func cliAgentName(spec runtime.ExecutionSpec) string {
	return spec.Target
}

func cancelEventForContext(err error) string {
	switch {
	case errors.Is(err, context.DeadlineExceeded):
		return "timeout"
	case errors.Is(err, context.Canceled):
		return "user-cancel"
	default:
		return "context-done"
	}
}

func normalizeSessionID(id string) string {
	if id == "" {
		return runtime.DefaultSessionID
	}
	return id
}

func closeProfileSession(_ context.Context, sess *profileSession) error {
	err := sess.closeFn()
	if sess.cmd != nil && sess.cmd.Process != nil {
		_ = sess.cmd.Process.Kill()
	}
	return err
}