// Package cli provides an Adapter that runs external CLI tools as an execution // adapter. Profiles (claude, antigravity, codex, opencode, cline) are defined by // edge configuration, pushed to the node in NodeConfigPayload, and selected as // the adapter execution target. 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" "strconv" "strings" "sync" "go.uber.org/zap" "iop/apps/node/internal/adapters/cli/status" "iop/apps/node/internal/runtime" "iop/packages/config" ) const Name = "cli" const ( modeCodexExec = "codex-exec" modeOpencodeSSE = "opencode-sse" modePersistentLazy = "persistent-lazy" ) type cliOutput struct { text string markerLine string } // sessionKey uniquely identifies a logical worker session. type sessionKey struct { target 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 opencodeSessions map[sessionKey]*opencodeSSESession logger *zap.Logger // StatusChecker overrides status.CheckUsage for testing. StatusChecker func(ctx context.Context, target 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), opencodeSessions: make(map[sessionKey]*opencodeSSESession), 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 shouldAutostartPersistentProfile(c.profiles[name]) { names = append(names, name) } } sort.Strings(names) for _, name := range names { profile := c.profiles[name] key := sessionKey{target: 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 target %q: %w", name, err) } c.mu.Lock() c.sessions[key] = sess c.mu.Unlock() c.logger.Info("cli adapter: persistent session started", zap.String("target", name)) } return nil } func shouldAutostartPersistentProfile(profile config.CLIProfileConf) bool { return profile.Persistent && profile.Mode != modeCodexExec && profile.Mode != modeOpencodeSSE && profile.Mode != modePersistentLazy } // 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.target, 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) opencodeCopy := make(map[sessionKey]*opencodeSSESession, len(c.opencodeSessions)) for key, sess := range c.opencodeSessions { opencodeCopy[key] = sess } c.opencodeSessions = make(map[sessionKey]*opencodeSSESession) c.mu.Unlock() for _, sess := range opencodeCopy { closeOpencodeSession(sess) } 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.target, key.sessionID, err) } } return firstErr } func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { target := cliTargetName(spec) profile, ok := c.profiles[target] if !ok { return fmt.Errorf("cli adapter: unknown target %q", target) } if profile.Mode == modeCodexExec { return c.executeCodexExec(ctx, spec, profile, sink) } if profile.Mode == modeOpencodeSSE { return c.executeOpencodeSSE(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) { switch req.Type { case runtime.CommandTypeUsageStatus: return c.handleUsageStatus(ctx, req) case runtime.CommandTypeSessionList: return c.handleSessionList(req), nil default: return runtime.CommandResponse{}, fmt.Errorf("cli adapter: unsupported command %q", req.Type) } } func (c *CLI) handleUsageStatus(ctx context.Context, req runtime.CommandRequest) (runtime.CommandResponse, error) { profile, ok := c.profiles[req.Target] if !ok { return runtime.CommandResponse{}, fmt.Errorf("cli adapter: unknown target %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 } func (c *CLI) handleSessionList(req runtime.CommandRequest) runtime.CommandResponse { c.mu.Lock() entries := make([]string, 0, len(c.sessions)+len(c.codexSessions)+len(c.opencodeSessions)) for k := range c.sessions { entries = append(entries, fmt.Sprintf("persistent:%s/%s", k.target, k.sessionID)) } for k := range c.codexSessions { entries = append(entries, fmt.Sprintf("codex-exec:%s/%s", k.target, k.sessionID)) } for k := range c.opencodeSessions { entries = append(entries, fmt.Sprintf("opencode-sse:%s/%s", k.target, k.sessionID)) } c.mu.Unlock() sort.Strings(entries) result := map[string]string{ "count": strconv.Itoa(len(entries)), "sessions": strings.Join(entries, ","), } return runtime.CommandResponse{ RequestID: req.RequestID, Type: req.Type, Adapter: req.Adapter, Target: req.Target, SessionID: req.SessionID, Result: result, } } // TerminateSession implements runtime.SessionTerminator. func (c *CLI) TerminateSession(_ context.Context, target, sessionID string) error { key := sessionKey{target: target, sessionID: normalizeSessionID(sessionID)} if profile, ok := c.profiles[target]; ok && profile.Mode == modeCodexExec { 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 target %q", key.sessionID, target) } return nil } if profile, ok := c.profiles[target]; ok && profile.Mode == modeOpencodeSSE { c.mu.Lock() sess, ok := c.opencodeSessions[key] if ok { delete(c.opencodeSessions, key) } c.mu.Unlock() if !ok { return fmt.Errorf("cli adapter: no session %q for target %q", key.sessionID, target) } closeOpencodeSession(sess) 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 target %q", key.sessionID, target) } return closeProfileSession(context.Background(), sess) } func cliTargetName(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 }