toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: CLI adapter with onshot/persistent mode support

Browse source 
- Add onshot.go and persistent.go for distinct CLI modes
- Update cli.go, factory_internal_test.go, cli_test.go
- Update configs/edge.yaml and packages/config
- Update READMEs for edge and node apps
This commit is contained in:
toki 2026-05-03 18:09:56 +09:00
parent 635ac41859
commit 24789c398b
10 changed files with 528 additions and 346 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
apps/edge/README.md
View file

@ -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] <Claude 응답 텍스트>
[node-local-node-message] <Codex 응답 텍스트>
```
프롬프트 없이 TCP/protobuf edge 서버만 실행하려면 기존처럼 `go run ./apps/edge/cmd/edge serve -c configs/edge.yaml`를 사용한다.

8
apps/node/README.md
View file

@ -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] <Claude 응답 텍스트>
[node-message] <Codex 응답 텍스트>
```
## 어댑터
@ -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

314
apps/node/internal/adapters/cli/cli.go
View file

@ -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)
}

42
apps/node/internal/adapters/cli/cli_test.go
View file

@ -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")

124
apps/node/internal/adapters/cli/oneshot.go Normal file
View file

@ -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)
}

233
apps/node/internal/adapters/cli/persistent.go Normal file
View file

@ -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
}
}
}

91
apps/node/internal/adapters/factory_internal_test.go
View file

@ -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")
}
}

32
configs/edge.yaml
View file

@ -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"

4
packages/config/config.go
View file

@ -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)
}

6
packages/config/config_test.go
View file

@ -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) {