diff --git a/apps/edge/README.md b/apps/edge/README.md index cb645c3..3544373 100644 --- a/apps/edge/README.md +++ b/apps/edge/README.md @@ -6,34 +6,36 @@ IOP 내부 TCP 프로토콜을 통해 node로 요청을 라우팅한다. **현재 상태: 초기 구현** node 등록/레지스트리/transport와 edge 콘솔 기반 수동 통신 테스트가 구현되어 있다. -## 원격 Claude CLI 수동 테스트 +## 원격 Codex CLI 수동 테스트 `bin/edge.sh`는 edge 서버를 열고 입력 프롬프트를 제공한다. -`bin/node.sh`는 원격 edge 주소로 접속해 등록한 뒤, edge에서 보낸 `RunRequest`를 cli/claude persistent terminal로 실행한다. +`bin/node.sh`는 원격 edge 주소로 접속해 등록한 뒤, edge에서 보낸 `RunRequest`를 cli/codex one-shot process로 실행한다. 테스트 파라미터는 `configs/edge.yaml`, `configs/node.yaml`에 하드코딩한다. -**전제 조건**: node 호스트의 PATH에 `claude` 명령이 있어야 한다. +**전제 조건**: node 호스트의 PATH에 `codex` 명령이 있어야 한다. + +같은 `cli` 어댑터 안에 `claude`, `gemini` profile도 포함할 수 있으며, 각각 `claude -p --dangerously-skip-permissions`, `gemini -p --approval-mode yolo`처럼 headless+bypass 조합으로 설정한다. 실행 순서: -1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. `console.adapter=cli`, `console.model=claude`가 기본값이다. +1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. `console.adapter=cli`, `console.model=codex`가 기본값이다. 2. node 호스트에서 `configs/node.yaml`의 `transport.edge_addr`를 edge 호스트 주소로, `transport.token`을 edge token과 같게 맞춘다. 3. edge 호스트의 `9090/tcp` 포트를 node 호스트에서 접근 가능하게 연다. 4. edge 호스트에서 `./bin/edge.sh` -5. node 호스트에서 `./bin/node.sh` — node 시작 직후 cli adapter가 `claude` persistent terminal process를 띄운다. +5. node 호스트에서 `./bin/node.sh` — node는 edge에서 받은 입력을 `codex exec --dangerously-bypass-approvals-and-sandbox` one-shot process로 실행한다. 6. edge 콘솔에서 `/nodes`로 node 등록을 확인한다. -7. edge 콘솔의 `edge>` 프롬프트에 메시지를 입력한다. 입력은 node의 Claude process stdin으로 전달된다. -8. `[node-{alias}-event]` 라인으로 실행 상태를, `[node-{alias}-message]` 라인으로 Claude 응답을 확인한다. +7. edge 콘솔의 `edge>` 프롬프트에 메시지를 입력한다. 입력은 node의 Codex process prompt 인자로 전달된다. +8. `[node-{alias}-event]` 라인으로 실행 상태를, `[node-{alias}-message]` 라인으로 Codex 응답을 확인한다. 9. edge 콘솔에서 `/exit` 또는 `quit`로 종료한다. 예상 출력: ```text edge> hello -[edge] sent run_id=manual-... node=local-node adapter=cli model=claude +[edge] sent run_id=manual-... node=local-node adapter=cli model=codex [node-local-node-event] start run_id=manual-... [node-local-node-event] complete run_id=manual-... detail="cli execution complete" -[node-local-node-message] +[node-local-node-message] ``` 프롬프트 없이 TCP/protobuf edge 서버만 실행하려면 기존처럼 `go run ./apps/edge/cmd/edge serve -c configs/edge.yaml`를 사용한다. diff --git a/apps/node/README.md b/apps/node/README.md index 53c588a..33cfe86 100644 --- a/apps/node/README.md +++ b/apps/node/README.md @@ -43,15 +43,15 @@ go build -o bin/node ./apps/node/cmd/node ./bin/node.sh ``` -node가 시작되면 cli adapter가 `claude` persistent terminal process를 띄운다. PATH에 `claude` 명령이 없으면 node 시작이 실패한다. +edge의 기본 설정은 cli adapter의 `codex` profile을 사용한다. PATH에 `codex` 명령이 있어야 하며, 각 실행 요청은 `codex exec --dangerously-bypass-approvals-and-sandbox` one-shot process로 처리된다. -edge에서 실행 요청을 받으면 node는 해당 입력을 Claude process의 stdin으로 전달하고, `response_idle_timeout_ms`(기본 1500ms) 동안 새 출력이 없으면 응답 경계로 처리해 complete 이벤트를 보낸다. +edge에서 실행 요청을 받으면 node는 해당 입력을 Codex process의 prompt 인자로 전달하고, process 종료 시 complete 이벤트를 보낸다. ```text [edge-message] hello [node-event] start run_id=manual-... [node-event] complete run_id=manual-... detail="cli execution complete" -[node-message] +[node-message] ``` ## 어댑터 @@ -63,6 +63,8 @@ edge에서 실행 요청을 받으면 node는 해당 입력을 Claude process의 | `vllm` | vLLM OpenAI-compatible API | TODO | | `cli` | claude/gemini/codex/opencode CLI 실행 | 구현 완료 | +`claude`, `gemini`, `codex`처럼 기본이 interactive TUI인 CLI는 이 파이프라인에서 non-interactive 모드로 설정해야 한다. 예: `claude -p --dangerously-skip-permissions`, `gemini -p --approval-mode yolo`, `codex exec --dangerously-bypass-approvals-and-sandbox`. + ## Transport - TCP 4-byte length-prefix framing diff --git a/apps/node/internal/adapters/cli/cli.go b/apps/node/internal/adapters/cli/cli.go index d9e410e..d08201f 100644 --- a/apps/node/internal/adapters/cli/cli.go +++ b/apps/node/internal/adapters/cli/cli.go @@ -1,20 +1,18 @@ // Package cli provides an Adapter that runs external CLI tools as inference // backends. Profiles (claude, gemini, codex, opencode) are configured in -// configs/node.yaml and select the command + args to execute. +// 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 ( - "bufio" "context" "fmt" "io" "os/exec" "sort" - "strings" "sync" - "time" - "github.com/creack/pty" "go.uber.org/zap" "iop/apps/node/internal/runtime" @@ -159,309 +157,3 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt } return c.executeOneShot(ctx, spec, profile, sink) } - -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) - 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(), - }) - } - - var errBuf strings.Builder - errScanner := bufio.NewScanner(stderr) - for errScanner.Scan() { - errBuf.WriteString(errScanner.Text()) - errBuf.WriteByte('\n') - } - - if err := cmd.Wait(); err != nil { - _ = sink.Emit(ctx, runtime.RuntimeEvent{ - RunID: spec.RunID, - Type: runtime.EventTypeError, - Error: fmt.Sprintf("command failed: %v — %s", err, errBuf.String()), - Timestamp: time.Now(), - }) - return fmt.Errorf("cli adapter: command exited with error: %w", err) - } - - 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 emitRuntimeError(ctx context.Context, sink runtime.EventSink, runID, msg string) error { - _ = sink.Emit(ctx, runtime.RuntimeEvent{ - RunID: runID, - Type: runtime.EventTypeError, - Error: msg, - Timestamp: time.Now(), - }) - return fmt.Errorf("cli adapter: %s", msg) -} - -func (c *CLI) executePersistent(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { - c.mu.Lock() - sess, ok := c.sessions[spec.Model] - c.mu.Unlock() - if !ok { - return fmt.Errorf("cli adapter: no persistent session for profile %q", spec.Model) - } - - sess.mu.Lock() - defer sess.mu.Unlock() - - prompt := extractPrompt(spec.Input) - idleTimeout := time.Duration(profile.ResponseIdleTimeoutMS) * time.Millisecond - if idleTimeout <= 0 { - idleTimeout = 1500 * time.Millisecond - } - - _ = sink.Emit(ctx, runtime.RuntimeEvent{ - RunID: spec.RunID, - Type: runtime.EventTypeStart, - Timestamp: time.Now(), - }) - - if _, err := fmt.Fprintf(sess.input, "%s\n", prompt); err != nil { - return emitRuntimeError(ctx, sink, spec.RunID, fmt.Sprintf("write prompt: %v", err)) - } - - // idle timer is nil until first output arrives (REVIEW_API-1) - var idleTimer *time.Timer - var idleC <-chan time.Time - defer func() { - if idleTimer != nil { - idleTimer.Stop() - } - }() - outputTokens := 0 - - for { - select { - case <-ctx.Done(): - // Only stop the specific session for this profile, not all sessions. - c.mu.Lock() - if s, ok := c.sessions[spec.Model]; ok && s == sess { - _ = s.closeFn() - if s.cmd != nil && s.cmd.Process != nil { - _ = s.cmd.Process.Kill() - } - delete(c.sessions, spec.Model) - } - c.mu.Unlock() - return ctx.Err() - case out, ok := <-sess.output: - if !ok { - // output channel closed means process exited — treat as failure (REVIEW_API-2) - return emitRuntimeError(ctx, sink, spec.RunID, "persistent session process exited unexpectedly") - } - outputTokens += len(strings.Fields(out.line)) - _ = sink.Emit(ctx, runtime.RuntimeEvent{ - RunID: spec.RunID, - Type: runtime.EventTypeDelta, - Delta: out.line + "\n", - Timestamp: time.Now(), - }) - // arm timer on first output; reset on subsequent outputs (REVIEW_API-1) - if idleTimer == nil { - idleTimer = time.NewTimer(idleTimeout) - idleC = idleTimer.C - } else { - if !idleTimer.Stop() { - select { - case <-idleTimer.C: - default: - } - } - idleTimer.Reset(idleTimeout) - } - case <-idleC: - 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(), - }) - case err := <-sess.done: - // persistent session process exit is always a failure (REVIEW_API-2) - msg := "persistent session process exited" - if err != nil { - msg = fmt.Sprintf("persistent session process exited: %v", err) - } - return emitRuntimeError(ctx, sink, spec.RunID, msg) - } - } -} - -func startProfileSession(_ context.Context, name string, profile config.CLIProfileConf, logger *zap.Logger) (*profileSession, error) { - if profile.Command == "" { - return nil, fmt.Errorf("profile %q has no command", name) - } - - outputCh := make(chan cliOutput, 1024) - doneCh := make(chan error, 1) - - var input io.Writer - var closeFn func() error - var cmd *exec.Cmd - - if profile.Terminal { - cmd = exec.Command(profile.Command, profile.Args...) - if len(profile.Env) > 0 { - cmd.Env = append(cmd.Environ(), profile.Env...) - } - ptmx, err := pty.Start(cmd) - if err != nil { - return nil, fmt.Errorf("pty start: %w", err) - } - input = ptmx - closeFn = ptmx.Close - go func() { - scanner := bufio.NewScanner(ptmx) - for scanner.Scan() { - line := strings.TrimRight(scanner.Text(), "\r") - outputCh <- cliOutput{line: line} - } - // send to buffered doneCh before closing outputCh so the startup - // drain's non-blocking doneCh check is race-free - doneCh <- cmd.Wait() - close(outputCh) - }() - } else { - cmd = exec.Command(profile.Command, profile.Args...) - if len(profile.Env) > 0 { - cmd.Env = append(cmd.Environ(), profile.Env...) - } - stdin, err := cmd.StdinPipe() - if err != nil { - return nil, fmt.Errorf("stdin pipe: %w", err) - } - stdout, err := cmd.StdoutPipe() - if err != nil { - _ = stdin.Close() - return nil, fmt.Errorf("stdout pipe: %w", err) - } - if err := cmd.Start(); err != nil { - _ = stdin.Close() - return nil, fmt.Errorf("start: %w", err) - } - input = stdin - closeFn = stdin.Close - go func() { - scanner := bufio.NewScanner(stdout) - for scanner.Scan() { - outputCh <- cliOutput{line: scanner.Text()} - } - doneCh <- cmd.Wait() - close(outputCh) - }() - } - - if profile.StartupIdleTimeoutMS > 0 { - drainUntilIdle(outputCh, time.Duration(profile.StartupIdleTimeoutMS)*time.Millisecond, logger, name) - } - - // check if process already exited during startup drain (REVIEW_API-2) - select { - case err := <-doneCh: - _ = closeFn() - if cmd.Process != nil { - _ = cmd.Process.Kill() - } - if err == nil { - return nil, fmt.Errorf("process exited during startup") - } - return nil, fmt.Errorf("process exited during startup: %w", err) - default: - } - - return &profileSession{ - name: name, - profile: profile, - cmd: cmd, - input: input, - output: outputCh, - done: doneCh, - closeFn: closeFn, - }, nil -} - -func drainUntilIdle(outputCh <-chan cliOutput, timeout time.Duration, logger *zap.Logger, name string) { - timer := time.NewTimer(timeout) - defer timer.Stop() - for { - select { - case _, ok := <-outputCh: - if !ok { - return - } - if !timer.Stop() { - select { - case <-timer.C: - default: - } - } - timer.Reset(timeout) - logger.Debug("cli adapter: startup drain", zap.String("profile", name)) - case <-timer.C: - return - } - } -} - -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) -} diff --git a/apps/node/internal/adapters/cli/cli_test.go b/apps/node/internal/adapters/cli/cli_test.go index ffafcab..c3eccd7 100644 --- a/apps/node/internal/adapters/cli/cli_test.go +++ b/apps/node/internal/adapters/cli/cli_test.go @@ -89,6 +89,48 @@ func TestCLIExecuteOneShotPassesPromptAsArg(t *testing.T) { } } +func TestCLIExecuteOneShotDrainsStderrConcurrently(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("sh not available on Windows") + } + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "noisy": { + Command: "sh", + Args: []string{ + "-c", + `i=0; while [ "$i" -lt 20000 ]; do echo "warning line $i" >&2; i=$((i+1)); done; printf "reply:%s\n" "$1"`, + "sh", + }, + }, + }, + } + c := New(cfg, zap.NewNop()) + sink := &fakeSink{} + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-noisy", + Model: "noisy", + Input: map[string]any{"prompt": "hello"}, + }, sink) + if err != nil { + t.Fatalf("execute: %v", err) + } + + var deltas []string + for _, e := range sink.Events() { + if e.Type == noderuntime.EventTypeDelta { + deltas = append(deltas, e.Delta) + } + } + if combined := strings.Join(deltas, ""); !strings.Contains(combined, "reply:hello") { + t.Fatalf("expected reply:hello in deltas, got %q", combined) + } +} + func TestCLIExecutePersistentWaitsForSlowFirstOutput(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("PTY not supported on Windows") diff --git a/apps/node/internal/adapters/cli/oneshot.go b/apps/node/internal/adapters/cli/oneshot.go new file mode 100644 index 0000000..0345f1d --- /dev/null +++ b/apps/node/internal/adapters/cli/oneshot.go @@ -0,0 +1,124 @@ +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) +} diff --git a/apps/node/internal/adapters/cli/persistent.go b/apps/node/internal/adapters/cli/persistent.go new file mode 100644 index 0000000..a940277 --- /dev/null +++ b/apps/node/internal/adapters/cli/persistent.go @@ -0,0 +1,233 @@ +package cli + +import ( + "bufio" + "context" + "fmt" + "io" + "os/exec" + "strings" + "time" + + "github.com/creack/pty" + "go.uber.org/zap" + + "iop/apps/node/internal/runtime" + "iop/packages/config" +) + +func emitRuntimeError(ctx context.Context, sink runtime.EventSink, runID, msg string) error { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: msg, + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: %s", msg) +} + +func (c *CLI) executePersistent(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { + c.mu.Lock() + sess, ok := c.sessions[spec.Model] + c.mu.Unlock() + if !ok { + return fmt.Errorf("cli adapter: no persistent session for profile %q", spec.Model) + } + + sess.mu.Lock() + defer sess.mu.Unlock() + + prompt := extractPrompt(spec.Input) + idleTimeout := time.Duration(profile.ResponseIdleTimeoutMS) * time.Millisecond + if idleTimeout <= 0 { + idleTimeout = 1500 * time.Millisecond + } + + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeStart, + Timestamp: time.Now(), + }) + + if _, err := fmt.Fprintf(sess.input, "%s\n", prompt); err != nil { + return emitRuntimeError(ctx, sink, spec.RunID, fmt.Sprintf("write prompt: %v", err)) + } + + var idleTimer *time.Timer + var idleC <-chan time.Time + defer func() { + if idleTimer != nil { + idleTimer.Stop() + } + }() + outputTokens := 0 + + for { + select { + case <-ctx.Done(): + c.mu.Lock() + if s, ok := c.sessions[spec.Model]; ok && s == sess { + _ = s.closeFn() + if s.cmd != nil && s.cmd.Process != nil { + _ = s.cmd.Process.Kill() + } + delete(c.sessions, spec.Model) + } + c.mu.Unlock() + return ctx.Err() + case out, ok := <-sess.output: + if !ok { + return emitRuntimeError(ctx, sink, spec.RunID, "persistent session process exited unexpectedly") + } + outputTokens += len(strings.Fields(out.line)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeDelta, + Delta: out.line + "\n", + Timestamp: time.Now(), + }) + if idleTimer == nil { + idleTimer = time.NewTimer(idleTimeout) + idleC = idleTimer.C + } else { + if !idleTimer.Stop() { + select { + case <-idleTimer.C: + default: + } + } + idleTimer.Reset(idleTimeout) + } + case <-idleC: + 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(), + }) + case err := <-sess.done: + msg := "persistent session process exited" + if err != nil { + msg = fmt.Sprintf("persistent session process exited: %v", err) + } + return emitRuntimeError(ctx, sink, spec.RunID, msg) + } + } +} + +func startProfileSession(_ context.Context, name string, profile config.CLIProfileConf, logger *zap.Logger) (*profileSession, error) { + if profile.Command == "" { + return nil, fmt.Errorf("profile %q has no command", name) + } + + outputCh := make(chan cliOutput, 1024) + doneCh := make(chan error, 1) + + var input io.Writer + var closeFn func() error + var cmd *exec.Cmd + + if profile.Terminal { + cmd = exec.Command(profile.Command, profile.Args...) + if len(profile.Env) > 0 { + cmd.Env = append(cmd.Environ(), profile.Env...) + } + ptmx, err := pty.Start(cmd) + if err != nil { + return nil, fmt.Errorf("pty start: %w", err) + } + input = ptmx + closeFn = ptmx.Close + go func() { + scanner := bufio.NewScanner(ptmx) + for scanner.Scan() { + line := strings.TrimRight(scanner.Text(), "\r") + outputCh <- cliOutput{line: line} + } + doneCh <- cmd.Wait() + close(outputCh) + }() + } else { + cmd = exec.Command(profile.Command, profile.Args...) + if len(profile.Env) > 0 { + cmd.Env = append(cmd.Environ(), profile.Env...) + } + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("stdin pipe: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = stdin.Close() + return nil, fmt.Errorf("stdout pipe: %w", err) + } + if err := cmd.Start(); err != nil { + _ = stdin.Close() + return nil, fmt.Errorf("start: %w", err) + } + input = stdin + closeFn = stdin.Close + go func() { + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + outputCh <- cliOutput{line: scanner.Text()} + } + doneCh <- cmd.Wait() + close(outputCh) + }() + } + + if profile.StartupIdleTimeoutMS > 0 { + drainUntilIdle(outputCh, time.Duration(profile.StartupIdleTimeoutMS)*time.Millisecond, logger, name) + } + + select { + case err := <-doneCh: + _ = closeFn() + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + if err == nil { + return nil, fmt.Errorf("process exited during startup") + } + return nil, fmt.Errorf("process exited during startup: %w", err) + default: + } + + return &profileSession{ + name: name, + profile: profile, + cmd: cmd, + input: input, + output: outputCh, + done: doneCh, + closeFn: closeFn, + }, nil +} + +func drainUntilIdle(outputCh <-chan cliOutput, timeout time.Duration, logger *zap.Logger, name string) { + timer := time.NewTimer(timeout) + defer timer.Stop() + for { + select { + case _, ok := <-outputCh: + if !ok { + return + } + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(timeout) + logger.Debug("cli adapter: startup drain", zap.String("profile", name)) + case <-timer.C: + return + } + } +} diff --git a/apps/node/internal/adapters/factory_internal_test.go b/apps/node/internal/adapters/factory_internal_test.go index e3d852a..000c1ca 100644 --- a/apps/node/internal/adapters/factory_internal_test.go +++ b/apps/node/internal/adapters/factory_internal_test.go @@ -10,13 +10,11 @@ func TestCLIConfFromStruct_ProfileRuntimeOptions(t *testing.T) { st, err := structpb.NewStruct(map[string]any{ "profiles": map[string]any{ "claude": map[string]any{ - "command": "claude", - "args": []any{}, - "env": []any{}, - "persistent": true, - "terminal": true, - "response_idle_timeout_ms": float64(1500), - "startup_idle_timeout_ms": float64(300), + "command": "claude", + "args": []any{"-p"}, + "env": []any{}, + "persistent": false, + "terminal": false, }, }, }) @@ -33,17 +31,80 @@ func TestCLIConfFromStruct_ProfileRuntimeOptions(t *testing.T) { if prof.Command != "claude" { t.Fatalf("command: got %q", prof.Command) } - if !prof.Persistent { - t.Fatal("expected persistent=true") + if len(prof.Args) != 1 || prof.Args[0] != "-p" { + t.Fatalf("args: got %#v", prof.Args) } - if !prof.Terminal { - t.Fatal("expected terminal=true") +} + +func TestCLIConfFromStruct_ClaudeHeadlessBypassProfile(t *testing.T) { + st, err := structpb.NewStruct(map[string]any{ + "profiles": map[string]any{ + "claude": map[string]any{ + "command": "claude", + "args": []any{"-p", "--dangerously-skip-permissions"}, + "env": []any{}, + "persistent": false, + "terminal": false, + }, + }, + }) + if err != nil { + t.Fatalf("structpb: %v", err) } - if prof.ResponseIdleTimeoutMS != 1500 { - t.Fatalf("response_idle_timeout_ms: got %d", prof.ResponseIdleTimeoutMS) + + cfg := cliConfFromStruct(st) + + prof, ok := cfg.Profiles["claude"] + if !ok { + t.Fatal("expected claude profile") } - if prof.StartupIdleTimeoutMS != 300 { - t.Fatalf("startup_idle_timeout_ms: got %d", prof.StartupIdleTimeoutMS) + if prof.Command != "claude" { + t.Fatalf("command: got %q", prof.Command) + } + if len(prof.Args) != 2 || prof.Args[0] != "-p" || prof.Args[1] != "--dangerously-skip-permissions" { + t.Fatalf("args: got %#v", prof.Args) + } + if prof.Persistent { + t.Fatal("expected persistent=false") + } + if prof.Terminal { + t.Fatal("expected terminal=false") + } +} + +func TestCLIConfFromStruct_GeminiHeadlessYoloProfile(t *testing.T) { + st, err := structpb.NewStruct(map[string]any{ + "profiles": map[string]any{ + "gemini": map[string]any{ + "command": "gemini", + "args": []any{"-p", "--approval-mode", "yolo"}, + "env": []any{}, + "persistent": false, + "terminal": false, + }, + }, + }) + if err != nil { + t.Fatalf("structpb: %v", err) + } + + cfg := cliConfFromStruct(st) + + prof, ok := cfg.Profiles["gemini"] + if !ok { + t.Fatal("expected gemini profile") + } + if prof.Command != "gemini" { + t.Fatalf("command: got %q", prof.Command) + } + if len(prof.Args) != 3 || prof.Args[0] != "-p" || prof.Args[1] != "--approval-mode" || prof.Args[2] != "yolo" { + t.Fatalf("args: got %#v", prof.Args) + } + if prof.Persistent { + t.Fatal("expected persistent=false") + } + if prof.Terminal { + t.Fatal("expected terminal=false") } } diff --git a/configs/edge.yaml b/configs/edge.yaml index 4fa093b..08499e7 100644 --- a/configs/edge.yaml +++ b/configs/edge.yaml @@ -13,7 +13,7 @@ metrics: console: adapter: "cli" - model: "claude" + model: "codex" timeout_sec: 120 nodes: @@ -31,12 +31,32 @@ nodes: profiles: claude: command: "claude" - args: [] + args: + - "-p" + - "--dangerously-skip-permissions" env: [] - persistent: true - terminal: true - response_idle_timeout_ms: 1500 - startup_idle_timeout_ms: 300 + persistent: false + terminal: false + gemini: + command: "gemini" + args: + - "-p" + - "--approval-mode" + - "yolo" + env: [] + persistent: false + terminal: false + codex: + command: "codex" + args: + - "exec" + - "--dangerously-bypass-approvals-and-sandbox" + - "--color" + - "never" + - "--skip-git-repo-check" + env: [] + persistent: false + terminal: false runtime: concurrency: 4 workspace_root: "/tmp/iop/workspace" diff --git a/packages/config/config.go b/packages/config/config.go index c8e8887..1f999ca 100644 --- a/packages/config/config.go +++ b/packages/config/config.go @@ -142,7 +142,7 @@ func setEdgeDefaults(v *viper.Viper) { v.SetDefault("logging.level", "info") v.SetDefault("metrics.port", 9092) v.SetDefault("tls.enabled", false) - v.SetDefault("console.adapter", "mock") - v.SetDefault("console.model", "mock-echo") + v.SetDefault("console.adapter", "cli") + v.SetDefault("console.model", "codex") v.SetDefault("console.timeout_sec", 120) } diff --git a/packages/config/config_test.go b/packages/config/config_test.go index 3f881c0..29311f8 100644 --- a/packages/config/config_test.go +++ b/packages/config/config_test.go @@ -22,6 +22,12 @@ func TestLoadEdge_ConsoleTimeoutDefault(t *testing.T) { if cfg.Console.TimeoutSec != 120 { t.Fatalf("expected default timeout_sec=120, got %d", cfg.Console.TimeoutSec) } + if cfg.Console.Adapter != "cli" { + t.Fatalf("expected default console.adapter=%q, got %q", "cli", cfg.Console.Adapter) + } + if cfg.Console.Model != "codex" { + t.Fatalf("expected default console.model=%q, got %q", "codex", cfg.Console.Model) + } } func TestLoadEdge_ConsoleTimeoutOverride(t *testing.T) {