// Package cli provides an Adapter that runs external CLI tools as inference // backends. 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" "fmt" "io" "os/exec" "sort" "sync" "go.uber.org/zap" "iop/apps/node/internal/runtime" "iop/packages/config" ) const Name = "cli" // known agents — must match CLI tool names var knownAgents = []string{"claude", "gemini", "codex", "opencode", "cline"} 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 } 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) { models := make([]string, 0, len(c.profiles)) for name := range c.profiles { models = append(models, name) } for _, agent := range knownAgents { if _, ok := c.profiles[agent]; !ok { models = append(models, agent) } } return runtime.Capabilities{ AdapterName: Name, Models: models, 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 && !isCodexExecProfile(c.profiles[name]) { 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.Persistent { if isCodexExecProfile(profile) { return c.executeCodexExec(ctx, spec, profile, sink) } return c.executePersistent(ctx, spec, profile, sink) } return c.executeOneShot(ctx, spec, profile, sink) } // 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 && isCodexExecProfile(profile) { 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.Model } 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 }