package cli import ( "bufio" "context" "errors" "fmt" "os" "os/exec" "strings" "time" "iop/apps/node/internal/runtime" "iop/packages/config" ) 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) var errBuf strings.Builder stderrDone := make(chan error, 1) go func() { errScanner := bufio.NewScanner(stderr) for errScanner.Scan() { errBuf.WriteString(errScanner.Text()) errBuf.WriteByte('\n') } stderrDone <- errScanner.Err() }() 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(), }) } if err := scanner.Err(); err != nil { _ = <-stderrDone _ = cmd.Wait() _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("read stdout: %v", err), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: read stdout: %w", err) } waitErr := cmd.Wait() stderrErr := <-stderrDone if waitErr != nil { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("command failed: %v — %s", waitErr, errBuf.String()), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: command exited with error: %w", waitErr) } if stderrErr != nil && !errors.Is(stderrErr, os.ErrClosed) { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("read stderr: %v", stderrErr), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: read stderr: %w", stderrErr) } 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 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) }