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

// 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"
	"os/exec"
	"sort"
	"strconv"
	"strings"
	"sync"

	"go.uber.org/zap"

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

const Name = "cli"

const (
	modeCodexExec      = "codex-exec"
	modeAntigravity    = "antigravity-print"
	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

	tailMu sync.Mutex
	tail   strings.Builder
	core   *terminalSessionCore
}

func (s *profileSession) appendTail(text string) {
	s.tailMu.Lock()
	defer s.tailMu.Unlock()
	appendBounded(&s.tail, text, 2048)
}

func (s *profileSession) getTail() string {
	if s.core != nil {
		return s.core.Snapshot().Tail
	}
	s.tailMu.Lock()
	defer s.tailMu.Unlock()
	return s.tail.String()
}

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

type antigravitySession struct {
	key            sessionKey
	conversationID string
	mu             sync.Mutex
}

type CLI struct {
	mu            sync.Mutex
	profiles      map[string]config.CLIProfileConf
	logger        *zap.Logger
	StatusChecker func(ctx context.Context, target string, profile config.CLIProfileConf) (*status.UsageStatus, error)

	oneShotExecutor     *oneshotExecutor
	persistentExecutor  *persistentExecutor
	codexExecutor       *codexExecutor
	antigravityExecutor *antigravityExecutor
	opencodeExecutor    *opencodeExecutor

	reporters []sessionReporter
}

type executor interface {
	Execute(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error
}

type sessionReporter interface {
	Sessions() []sessionListEntry
	Terminate(ctx context.Context, target, sessionID string) (bool, error)
	Stop(ctx context.Context) error
}

type oneshotExecutor struct {
	cli *CLI
}

func (e *oneshotExecutor) Execute(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error {
	return e.cli.executeOneShot(ctx, spec, profile, sink)
}

type persistentExecutor struct {
	cli      *CLI
	mu       sync.Mutex
	sessions map[sessionKey]*profileSession
}

type codexExecutor struct {
	cli      *CLI
	mu       sync.Mutex
	sessions map[sessionKey]*codexExecSession
}

type antigravityExecutor struct {
	cli      *CLI
	mu       sync.Mutex
	sessions map[sessionKey]*antigravitySession
}

type opencodeExecutor struct {
	cli      *CLI
	mu       sync.Mutex
	sessions map[sessionKey]*opencodeSSESession
}

func New(cfg config.CLIConf, logger *zap.Logger) *CLI {
	c := &CLI{
		profiles: cfg.Profiles,
		logger:   logger,
	}
	c.oneShotExecutor = &oneshotExecutor{cli: c}
	c.persistentExecutor = &persistentExecutor{
		cli:      c,
		sessions: make(map[sessionKey]*profileSession),
	}
	c.codexExecutor = &codexExecutor{
		cli:      c,
		sessions: make(map[sessionKey]*codexExecSession),
	}
	c.antigravityExecutor = &antigravityExecutor{
		cli:      c,
		sessions: make(map[sessionKey]*antigravitySession),
	}
	c.opencodeExecutor = &opencodeExecutor{
		cli:      c,
		sessions: make(map[sessionKey]*opencodeSSESession),
	}
	c.reporters = []sessionReporter{
		c.persistentExecutor,
		c.codexExecutor,
		c.antigravityExecutor,
		c.opencodeExecutor,
	}
	return c
}

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.
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.persistentExecutor.mu.Lock()
			_ = c.persistentExecutor.stopAllSessions(context.Background())
			c.persistentExecutor.mu.Unlock()
			return fmt.Errorf("cli adapter: start target %q: %w", name, err)
		}
		c.persistentExecutor.mu.Lock()
		c.persistentExecutor.sessions[key] = sess
		c.persistentExecutor.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 != modeAntigravity &&
		profile.Mode != modeOpencodeSSE &&
		profile.Mode != modePersistentLazy
}

// Stop stops all logical sessions. Errors are combined by reporting only the first.
func (c *CLI) Stop(ctx context.Context) error {
	var firstErr error
	for _, reporter := range c.reporters {
		if err := reporter.Stop(ctx); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

func (c *CLI) executorFor(profile config.CLIProfileConf) executor {
	switch profile.Mode {
	case modeCodexExec:
		return c.codexExecutor
	case modeAntigravity:
		return c.antigravityExecutor
	case modeOpencodeSSE:
		return c.opencodeExecutor
	default:
		if profile.Persistent {
			return c.persistentExecutor
		}
		return c.oneShotExecutor
	}
}

func (c *CLI) sessionReporterFor(profile config.CLIProfileConf) sessionReporter {
	switch profile.Mode {
	case modeCodexExec:
		return c.codexExecutor
	case modeAntigravity:
		return c.antigravityExecutor
	case modeOpencodeSSE:
		return c.opencodeExecutor
	default:
		if profile.Persistent {
			return c.persistentExecutor
		}
		return nil
	}
}

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)
	}
	return c.executorFor(profile).Execute(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
	}
	if st == nil {
		return runtime.CommandResponse{}, fmt.Errorf("cli adapter: status checker returned nil result for target %q", req.Target)
	}
	runtimeStatus := st.ToRuntime()
	if runtimeStatus.Metadata == nil {
		runtimeStatus.Metadata = make(map[string]string)
	}
	annotateUsageParseStatus(runtimeStatus)
	return runtime.CommandResponse{
		RequestID:   req.RequestID,
		Type:        req.Type,
		Adapter:     req.Adapter,
		Target:      req.Target,
		SessionID:   req.SessionID,
		UsageStatus: runtimeStatus,
	}, nil
}

// annotateUsageParseStatus sets parse_status metadata only when no structured data is present.
// Structured data means any of: daily/weekly limit fields set, or pre-existing metadata entries.
// - raw-only: RawOutput present but no structured data
// - empty:    RawOutput absent and no structured data
// - (no annotation) if structured data exists (parsed fields or metadata-only result)
func annotateUsageParseStatus(s *runtime.AgentUsageStatus) {
	hasFields := s.DailyLimit != "" || s.DailyResetTime != "" ||
		s.WeeklyLimit != "" || s.WeeklyResetTime != ""
	if hasFields || len(s.Metadata) > 0 {
		return
	}
	if s.RawOutput == "" {
		s.Metadata["parse_status"] = "empty"
	} else {
		s.Metadata["parse_status"] = "raw_only"
	}
}

// sessionListEntry holds a typed snapshot of a single logical session for SESSION_LIST.
type sessionListEntry struct {
	mode      string
	target    string
	sessionID string
}

func (e sessionListEntry) label() string {
	return e.mode + ":" + e.target + "/" + e.sessionID
}

func (c *CLI) handleSessionList(req runtime.CommandRequest) runtime.CommandResponse {
	var snaps []sessionListEntry
	for _, r := range c.reporters {
		snaps = append(snaps, r.Sessions()...)
	}
	sort.Slice(snaps, func(i, j int) bool { return snaps[i].label() < snaps[j].label() })
	labels := make([]string, len(snaps))
	for i, s := range snaps {
		labels[i] = s.label()
	}
	result := map[string]string{
		"count":    strconv.Itoa(len(snaps)),
		"sessions": strings.Join(labels, ","),
	}
	for i, s := range snaps {
		prefix := fmt.Sprintf("session.%d.", i)
		result[prefix+"label"] = s.label()
		result[prefix+"mode"] = s.mode
		result[prefix+"target"] = s.target
		result[prefix+"session_id"] = s.sessionID
	}
	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(ctx context.Context, target, sessionID string) error {
	var reporter sessionReporter
	if profile, ok := c.profiles[target]; ok {
		reporter = c.sessionReporterFor(profile)
	}
	if reporter == nil {
		reporter = c.persistentExecutor
	}
	terminated, err := reporter.Terminate(ctx, target, sessionID)
	if err != nil {
		return err
	}
	if !terminated {
		return fmt.Errorf("cli adapter: no session %q for target %q", normalizeSessionID(sessionID), target)
	}
	return nil
}

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 isAlreadyClosedError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, os.ErrClosed) {
		return true
	}
	errStr := err.Error()
	return strings.Contains(errStr, "file already closed") ||
		strings.Contains(errStr, "use of closed file")
}

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