From 157a8a707643b883fe1642c2a40e44639cf891d6 Mon Sep 17 00:00:00 2001 From: toki Date: Mon, 4 May 2026 07:55:08 +0900 Subject: [PATCH] feat: add console events, codex execution support and update config - add console_events.go for edge console event handling - add codex_exec.go for CLI codex execution - add codex_exec_test.go for persistent CLI tests - update console.go and oneshot.go with new features - update config and edge.yaml for new settings - update README files for edge and node apps --- apps/edge/README.md | 10 +- apps/edge/cmd/edge/console.go | 39 +-- apps/edge/cmd/edge/console_events.go | 156 ++++++++++ apps/edge/cmd/edge/console_test.go | 63 ++++ apps/node/README.md | 8 +- apps/node/internal/adapters/cli/cli.go | 41 ++- apps/node/internal/adapters/cli/codex_exec.go | 120 ++++++++ apps/node/internal/adapters/cli/oneshot.go | 281 ++++++++++++++++-- .../internal/adapters/cli/oneshot/cli_test.go | 153 ++++++++++ .../cli/persistent/codex_exec_test.go | 225 ++++++++++++++ configs/edge.yaml | 18 +- packages/config/config.go | 7 +- packages/config/config_test.go | 4 +- 13 files changed, 1051 insertions(+), 74 deletions(-) create mode 100644 apps/edge/cmd/edge/console_events.go create mode 100644 apps/node/internal/adapters/cli/codex_exec.go create mode 100644 apps/node/internal/adapters/cli/persistent/codex_exec_test.go diff --git a/apps/edge/README.md b/apps/edge/README.md index 525cc58..a1fe49b 100644 --- a/apps/edge/README.md +++ b/apps/edge/README.md @@ -12,17 +12,17 @@ node 등록/레지스트리/transport와 edge 콘솔 기반 수동 통신 테스 `bin/node.sh`는 원격 edge 주소로 접속해 등록한 뒤, edge에서 보낸 `RunRequest`를 cli/codex one-shot process로 실행한다. 테스트 파라미터는 `configs/edge.yaml`, `configs/node.yaml`에 하드코딩한다. -**전제 조건**: node 호스트의 PATH에 `codex` 명령이 있어야 한다. +**전제 조건**: node 호스트의 PATH에 `claude` 명령이 있어야 한다. -같은 `cli` 어댑터 안에 `claude`, `gemini` profile도 포함할 수 있으며, 각각 `claude -p --dangerously-skip-permissions`, `gemini -p --approval-mode yolo`처럼 headless+bypass 조합으로 설정한다. +같은 `cli` 어댑터 안에 `gemini`, `codex` profile도 포함할 수 있으며, 각각 `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`처럼 headless+bypass 조합으로 설정한다. Claude는 `claude -p --dangerously-skip-permissions --output-format stream-json --include-partial-messages --verbose ` 형태로 호출되어 token-level streaming이 활성화된다. 실행 순서: -1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. `console.adapter=cli`, `console.model=codex`, `console.session_id=default`가 기본값이다. +1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. `console.adapter=cli`, `console.model=claude`, `console.session_id=default`가 기본값이다. 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는 edge에서 받은 입력을 `codex exec --dangerously-bypass-approvals-and-sandbox` persistent process로 실행한다. +5. node 호스트에서 `./bin/node.sh` — node는 edge에서 받은 입력을 `claude -p --dangerously-skip-permissions --output-format stream-json ... ` one-shot process로 실행한다. 6. edge 콘솔에서 `/nodes`로 node 등록을 확인한다. 7. edge 콘솔의 `edge>` 프롬프트에 메시지를 입력한다. 8. `[node-{alias}-event]` 라인으로 실행 상태를, `[node-{alias}-message]` 라인으로 Codex 응답을 확인한다. @@ -32,7 +32,7 @@ node 등록/레지스트리/transport와 edge 콘솔 기반 수동 통신 테스 ```text edge> hello -[edge] sent run_id=manual-... node=local-node adapter=cli model=codex session=default background=false +[edge] sent run_id=manual-... node=local-node adapter=cli model=claude session=default background=false [node-local-node-event] start run_id=manual-... [node-local-node-event] complete run_id=manual-... detail="cli execution complete" [node-local-node-message] diff --git a/apps/edge/cmd/edge/console.go b/apps/edge/cmd/edge/console.go index a79ad02..54ac0a0 100644 --- a/apps/edge/cmd/edge/console.go +++ b/apps/edge/cmd/edge/console.go @@ -35,14 +35,8 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i if err != nil { return err } - events := make(chan *iop.RunEvent, 256) - server.SetRunEventHandler(func(event *iop.RunEvent) { - select { - case events <- event: - default: - logger.Warn("run event dropped", zap.String("run_id", event.GetRunId())) - } - }) + events := newConsoleEventRouter(out, logger) + server.SetRunEventHandler(events.Handle) if err := server.Start(ctx); err != nil { return err @@ -160,7 +154,7 @@ func buildRunRequest(adapter, model, sessionID string, background bool, timeoutS return req, runID, nil } -func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-chan *iop.RunEvent, out io.Writer, adapter, model, sessionID string, background bool, timeoutSec int, message string) error { +func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *consoleEventRouter, out io.Writer, adapter, model, sessionID string, background bool, timeoutSec int, message string) error { entry, err := registry.Pick() if err != nil { return err @@ -171,6 +165,13 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-c return err } + var runEvents <-chan *iop.RunEvent + var unregister func() + if !background { + runEvents, unregister = events.Register(runID) + defer unregister() + } + nodeAlias := entry.Alias fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s model=%s session=%s background=%v\n", runID, nodeAlias, adapter, model, req.GetSessionId(), background) @@ -185,7 +186,7 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-c timer := time.NewTimer(time.Duration(timeoutSec+5) * time.Second) defer timer.Stop() - var response strings.Builder + response := newConsoleResponseStream(out, fmt.Sprintf("[node-%s-message] ", nodeAlias)) for { select { @@ -193,7 +194,7 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-c return ctx.Err() case <-timer.C: return fmt.Errorf("timed out waiting for node response") - case event := <-events: + case event := <-runEvents: if event.GetRunId() != runID { continue } @@ -201,17 +202,18 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events <-c case "start": fmt.Fprintf(out, "[node-%s-event] start run_id=%s\n", nodeAlias, runID) case "delta": - response.WriteString(event.GetDelta()) + response.Write(event.GetDelta()) case "complete": + response.Finish() fmt.Fprintf(out, "[node-%s-event] complete run_id=%s detail=%q\n", nodeAlias, runID, event.GetMessage()) - printNodeMessage(out, nodeAlias, response.String()) return nil case "cancelled": + response.FinishIfStarted() fmt.Fprintf(out, "[node-%s-event] cancelled run_id=%s\n", nodeAlias, runID) return nil case "error": + response.FinishIfStarted() fmt.Fprintf(out, "[node-%s-event] error run_id=%s detail=%q\n", nodeAlias, runID, event.GetError()) - printNodeMessage(out, nodeAlias, response.String()) return fmt.Errorf("node reported error") default: fmt.Fprintf(out, "[node-%s-event] %s run_id=%s detail=%q\n", nodeAlias, event.GetType(), runID, event.GetMessage()) @@ -234,15 +236,6 @@ func sendTerminateSession(ctx context.Context, registry *edgenode.Registry, adap return entry.Client.Send(req) } -func printNodeMessage(out io.Writer, alias, message string) { - message = strings.TrimSpace(message) - if message == "" { - fmt.Fprintf(out, "[node-%s-message] \n", alias) - return - } - fmt.Fprintf(out, "[node-%s-message] %s\n", alias, message) -} - func normalizeConsoleSessionID(id string) string { if id == "" { return "default" diff --git a/apps/edge/cmd/edge/console_events.go b/apps/edge/cmd/edge/console_events.go new file mode 100644 index 0000000..cf41c16 --- /dev/null +++ b/apps/edge/cmd/edge/console_events.go @@ -0,0 +1,156 @@ +package main + +import ( + "fmt" + "io" + "strings" + "sync" + + "go.uber.org/zap" + + iop "iop/proto/gen/iop" +) + +type consoleEventRouter struct { + mu sync.Mutex + out io.Writer + logger *zap.Logger + waiters map[string]chan *iop.RunEvent + streams map[string]*consoleResponseStream +} + +func newConsoleEventRouter(out io.Writer, logger *zap.Logger) *consoleEventRouter { + return &consoleEventRouter{ + out: out, + logger: logger, + waiters: make(map[string]chan *iop.RunEvent), + streams: make(map[string]*consoleResponseStream), + } +} + +func (r *consoleEventRouter) Register(runID string) (<-chan *iop.RunEvent, func()) { + ch := make(chan *iop.RunEvent, 4096) + r.mu.Lock() + r.waiters[runID] = ch + r.mu.Unlock() + + return ch, func() { + r.mu.Lock() + if current, ok := r.waiters[runID]; ok && current == ch { + delete(r.waiters, runID) + close(ch) + } + r.mu.Unlock() + } +} + +func (r *consoleEventRouter) Handle(event *iop.RunEvent) { + runID := event.GetRunId() + + r.mu.Lock() + if waiter, ok := r.waiters[runID]; ok { + select { + case waiter <- event: + default: + if r.logger != nil { + r.logger.Warn("console run event dropped", zap.String("run_id", runID), zap.String("type", event.GetType())) + } + } + r.mu.Unlock() + return + } + r.printAsyncLocked(event) + r.mu.Unlock() +} + +func (r *consoleEventRouter) printAsyncLocked(event *iop.RunEvent) { + if r.out == nil { + return + } + + runID := event.GetRunId() + switch event.GetType() { + case "start": + fmt.Fprintf(r.out, "[node-event] start run_id=%s session=%s background=%v\n", runID, event.GetSessionId(), event.GetBackground()) + case "delta": + r.responseStream(runID).Write(event.GetDelta()) + case "complete": + r.finishStream(runID) + fmt.Fprintf(r.out, "[node-event] complete run_id=%s detail=%q\n", runID, event.GetMessage()) + case "cancelled": + r.finishStreamIfStarted(runID) + fmt.Fprintf(r.out, "[node-event] cancelled run_id=%s\n", runID) + case "error": + r.finishStreamIfStarted(runID) + fmt.Fprintf(r.out, "[node-event] error run_id=%s detail=%q\n", runID, event.GetError()) + default: + fmt.Fprintf(r.out, "[node-event] %s run_id=%s detail=%q\n", event.GetType(), runID, event.GetMessage()) + } +} + +func (r *consoleEventRouter) responseStream(runID string) *consoleResponseStream { + if s, ok := r.streams[runID]; ok { + return s + } + s := newConsoleResponseStream(r.out, "[node-message] ") + r.streams[runID] = s + return s +} + +func (r *consoleEventRouter) finishStream(runID string) { + if s, ok := r.streams[runID]; ok { + s.Finish() + delete(r.streams, runID) + return + } + newConsoleResponseStream(r.out, "[node-message] ").Finish() +} + +func (r *consoleEventRouter) finishStreamIfStarted(runID string) { + if s, ok := r.streams[runID]; ok { + s.FinishIfStarted() + delete(r.streams, runID) + } +} + +type consoleResponseStream struct { + out io.Writer + prefix string + started bool + endedWithNewline bool +} + +func newConsoleResponseStream(out io.Writer, prefix string) *consoleResponseStream { + return &consoleResponseStream{out: out, prefix: prefix, endedWithNewline: true} +} + +func (s *consoleResponseStream) Write(delta string) { + if s.out == nil || delta == "" { + return + } + if !s.started { + fmt.Fprint(s.out, s.prefix) + s.started = true + } + fmt.Fprint(s.out, delta) + s.endedWithNewline = strings.HasSuffix(delta, "\n") +} + +func (s *consoleResponseStream) Finish() { + if s.out == nil { + return + } + if !s.started { + fmt.Fprintf(s.out, "%s\n", s.prefix) + return + } + if !s.endedWithNewline { + fmt.Fprintln(s.out) + } +} + +func (s *consoleResponseStream) FinishIfStarted() { + if s.started { + s.Finish() + } +} diff --git a/apps/edge/cmd/edge/console_test.go b/apps/edge/cmd/edge/console_test.go index 8a0c229..bd7f945 100644 --- a/apps/edge/cmd/edge/console_test.go +++ b/apps/edge/cmd/edge/console_test.go @@ -1,6 +1,8 @@ package main import ( + "bytes" + "strings" "testing" iop "iop/proto/gen/iop" @@ -59,3 +61,64 @@ func TestNormalizeConsoleSessionID(t *testing.T) { t.Errorf("non-empty: got %q want %q", got, "my-session") } } + +func TestConsoleEventRouterRoutesRegisteredRun(t *testing.T) { + var out bytes.Buffer + router := newConsoleEventRouter(&out, nil) + events, unregister := router.Register("run-1") + defer unregister() + + router.Handle(&iop.RunEvent{RunId: "run-1", Type: "start"}) + + select { + case event := <-events: + if event.GetRunId() != "run-1" { + t.Fatalf("RunId: got %q want run-1", event.GetRunId()) + } + default: + t.Fatal("expected registered run event") + } + if out.Len() != 0 { + t.Fatalf("registered foreground event should not be printed asynchronously: %q", out.String()) + } +} + +func TestConsoleEventRouterPrintsUnregisteredRun(t *testing.T) { + var out bytes.Buffer + router := newConsoleEventRouter(&out, nil) + + router.Handle(&iop.RunEvent{RunId: "run-bg", Type: "start", SessionId: "s1", Background: true}) + router.Handle(&iop.RunEvent{RunId: "run-bg", Type: "delta", Delta: "hello"}) + if got := out.String(); !strings.Contains(got, "[node-message] hello") { + t.Fatalf("expected delta to be printed before completion, got:\n%s", got) + } + router.Handle(&iop.RunEvent{RunId: "run-bg", Type: "delta", Delta: " world\n"}) + router.Handle(&iop.RunEvent{RunId: "run-bg", Type: "complete", Message: "done"}) + + got := out.String() + for _, want := range []string{ + "[node-event] start run_id=run-bg session=s1 background=true", + "[node-event] complete run_id=run-bg detail=\"done\"", + "[node-message] hello world", + } { + if !strings.Contains(got, want) { + t.Fatalf("expected output to contain %q, got:\n%s", want, got) + } + } +} + +func TestConsoleResponseStreamWritesBeforeFinish(t *testing.T) { + var out bytes.Buffer + stream := newConsoleResponseStream(&out, "[node-test-message] ") + + stream.Write("hello") + if got := out.String(); got != "[node-test-message] hello" { + t.Fatalf("Write should print immediately, got %q", got) + } + + stream.Write(" world") + stream.Finish() + if got := out.String(); got != "[node-test-message] hello world\n" { + t.Fatalf("Finish should close the line, got %q", got) + } +} diff --git a/apps/node/README.md b/apps/node/README.md index 14c8103..7ce2b96 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 ``` -edge의 기본 설정은 cli adapter의 `codex` profile을 사용한다. PATH에 `codex` 명령이 있어야 하며, 각 실행 요청은 `codex exec --dangerously-bypass-approvals-and-sandbox` one-shot process로 처리된다. +edge의 기본 설정은 cli adapter의 `claude` profile을 사용한다. PATH에 `claude` 명령이 있어야 하며, 각 실행 요청은 `claude -p --dangerously-skip-permissions --output-format stream-json --include-partial-messages --verbose ` one-shot process로 처리된다 (token-level streaming). -edge에서 실행 요청을 받으면 node는 해당 입력을 Codex process의 prompt 인자로 전달하고, complete 이벤트를 보낸다. +edge에서 실행 요청을 받으면 node는 해당 입력을 Claude process의 prompt 인자로 전달하고, stream-json delta를 토큰 단위로 emit한다. ```text [edge-message] hello [node-event] start run_id=manual-... [node-event] complete run_id=manual-... detail="cli execution complete" -[node-message] +[node-message] ``` ## Logical Session (transport 1개 · session 여러 개) @@ -83,7 +83,7 @@ edge-node transport 연결은 **호스트당 1개**를 유지한다. 그 연결 | `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`. +`claude`, `gemini`, `codex`처럼 기본이 interactive TUI인 CLI는 이 파이프라인에서 non-interactive 모드로 설정해야 한다. 예: `claude -p --dangerously-skip-permissions`, `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`. ## Transport diff --git a/apps/node/internal/adapters/cli/cli.go b/apps/node/internal/adapters/cli/cli.go index 1fb003f..c8883cd 100644 --- a/apps/node/internal/adapters/cli/cli.go +++ b/apps/node/internal/adapters/cli/cli.go @@ -46,18 +46,26 @@ type profileSession struct { 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 - logger *zap.Logger + mu sync.Mutex + profiles map[string]config.CLIProfileConf + sessions map[sessionKey]*profileSession + codexSessions map[sessionKey]*codexExecSession + logger *zap.Logger } func New(cfg config.CLIConf, logger *zap.Logger) *CLI { return &CLI{ - profiles: cfg.Profiles, - sessions: make(map[sessionKey]*profileSession), - logger: logger, + profiles: cfg.Profiles, + sessions: make(map[sessionKey]*profileSession), + codexSessions: make(map[sessionKey]*codexExecSession), + logger: logger, } } @@ -85,7 +93,7 @@ func (c *CLI) Capabilities(_ context.Context) (runtime.Capabilities, error) { func (c *CLI) Start(ctx context.Context) error { names := make([]string, 0, len(c.profiles)) for name := range c.profiles { - if c.profiles[name].Persistent { + if c.profiles[name].Persistent && !isCodexExecProfile(c.profiles[name]) { names = append(names, name) } } @@ -131,6 +139,7 @@ func (c *CLI) Stop(_ context.Context) error { sessionsCopy[key] = sess } c.sessions = make(map[sessionKey]*profileSession) + c.codexSessions = make(map[sessionKey]*codexExecSession) c.mu.Unlock() var firstErr error @@ -148,6 +157,9 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt return fmt.Errorf("cli adapter: unknown profile %q", spec.Model) } if profile.Persistent { + if isCodexExecProfile(profile) { + return c.executeCodexExec(ctx, spec, profile, sink) + } return c.executePersistent(ctx, spec, profile, sink) } return c.executeOneShot(ctx, spec, profile, sink) @@ -156,6 +168,19 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt // TerminateSession implements runtime.SessionTerminator. func (c *CLI) TerminateSession(_ context.Context, model, sessionID string) error { key := sessionKey{model: model, sessionID: normalizeSessionID(sessionID)} + if profile, ok := c.profiles[model]; ok && isCodexExecProfile(profile) { + 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 profile %q", key.sessionID, model) + } + return nil + } + c.mu.Lock() sess, ok := c.sessions[key] if ok { diff --git a/apps/node/internal/adapters/cli/codex_exec.go b/apps/node/internal/adapters/cli/codex_exec.go new file mode 100644 index 0000000..2d4f46c --- /dev/null +++ b/apps/node/internal/adapters/cli/codex_exec.go @@ -0,0 +1,120 @@ +package cli + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "regexp" + "strings" + + "iop/apps/node/internal/runtime" + "iop/packages/config" +) + +var ( + codexSessionIDPattern = regexp.MustCompile(`(?mi)^session id:\s*(\S+)\s*$`) + codexThreadStartedRegex = regexp.MustCompile(`\{[^{}]*"type"\s*:\s*"thread\.started"[^{}]*\}`) +) + +func isCodexExecProfile(profile config.CLIProfileConf) bool { + return filepath.Base(profile.Command) == "codex" && len(profile.Args) > 0 && profile.Args[0] == "exec" +} + +func (c *CLI) executeCodexExec(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { + sess, err := c.resolveCodexExecSession(spec) + if err != nil { + return err + } + + sess.mu.Lock() + defer sess.mu.Unlock() + + prompt := extractPrompt(spec.Input) + args := codexExecArgs(profile.Args, sess.externalID, prompt) + output, err := c.executeCommand(ctx, spec, profile, args, prompt, sink) + if err != nil { + return err + } + + if externalID := parseCodexSessionID(output); externalID != "" { + sess.externalID = externalID + } + if sess.externalID == "" { + return fmt.Errorf("cli adapter: codex exec did not report a session id") + } + return nil +} + +func (c *CLI) resolveCodexExecSession(spec runtime.ExecutionSpec) (*codexExecSession, error) { + key := sessionKey{model: spec.Model, sessionID: normalizeSessionID(spec.SessionID)} + + c.mu.Lock() + defer c.mu.Unlock() + + if sess, ok := c.codexSessions[key]; ok { + return sess, nil + } + if spec.SessionMode == runtime.SessionModeRequireExisting { + return nil, fmt.Errorf("cli adapter: no persistent session for profile %q session %q", spec.Model, key.sessionID) + } + sess := &codexExecSession{key: key} + c.codexSessions[key] = sess + return sess, nil +} + +func codexExecArgs(configured []string, externalSessionID, prompt string) []string { + if externalSessionID == "" { + args := append([]string{}, configured...) + return append(args, prompt) + } + + args := []string{"exec", "resume"} + args = append(args, codexResumeOptions(configured)...) + args = append(args, externalSessionID, prompt) + return args +} + +func codexResumeOptions(configured []string) []string { + if len(configured) > 0 && configured[0] == "exec" { + configured = configured[1:] + } + + out := make([]string, 0, len(configured)) + for i := 0; i < len(configured); i++ { + arg := configured[i] + switch { + case arg == "--color" || arg == "-C" || arg == "--cd" || arg == "-s" || arg == "--sandbox" || + arg == "--profile" || arg == "-p" || arg == "--local-provider" || arg == "--add-dir" || + arg == "--output-schema": + if i+1 < len(configured) { + i++ + } + continue + case strings.HasPrefix(arg, "--color=") || strings.HasPrefix(arg, "--cd=") || + strings.HasPrefix(arg, "--sandbox=") || strings.HasPrefix(arg, "--profile=") || + strings.HasPrefix(arg, "--local-provider=") || strings.HasPrefix(arg, "--add-dir=") || + strings.HasPrefix(arg, "--output-schema="): + continue + default: + out = append(out, arg) + } + } + return out +} + +func parseCodexSessionID(output string) string { + if matches := codexSessionIDPattern.FindStringSubmatch(output); len(matches) >= 2 { + return matches[1] + } + for _, raw := range codexThreadStartedRegex.FindAllString(output, -1) { + var ev struct { + Type string `json:"type"` + ThreadID string `json:"thread_id"` + } + if err := json.Unmarshal([]byte(raw), &ev); err == nil && ev.Type == "thread.started" && ev.ThreadID != "" { + return ev.ThreadID + } + } + return "" +} diff --git a/apps/node/internal/adapters/cli/oneshot.go b/apps/node/internal/adapters/cli/oneshot.go index 0345f1d..fc848f0 100644 --- a/apps/node/internal/adapters/cli/oneshot.go +++ b/apps/node/internal/adapters/cli/oneshot.go @@ -3,8 +3,10 @@ package cli import ( "bufio" "context" + "encoding/json" "errors" "fmt" + "io" "os" "os/exec" "strings" @@ -17,6 +19,11 @@ import ( 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) + _, err := c.executeCommand(ctx, spec, profile, args, prompt, sink) + return err +} + +func (c *CLI) executeCommand(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, args []string, prompt string, sink runtime.EventSink) (string, error) { cmd := exec.CommandContext(ctx, profile.Command, args...) if len(profile.Env) > 0 { @@ -25,15 +32,15 @@ func (c *CLI) executeOneShot(ctx context.Context, spec runtime.ExecutionSpec, pr stdout, err := cmd.StdoutPipe() if err != nil { - return fmt.Errorf("cli adapter: stdout pipe: %w", err) + 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) + 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) + return "", fmt.Errorf("cli adapter: start %q: %w", profile.Command, err) } _ = sink.Emit(ctx, runtime.RuntimeEvent{ @@ -42,52 +49,62 @@ func (c *CLI) executeOneShot(ctx context.Context, spec runtime.ExecutionSpec, pr Timestamp: time.Now(), }) - scanner := bufio.NewScanner(stdout) var errBuf strings.Builder + var outBuf strings.Builder + combinedOutput := func() string { + return outBuf.String() + errBuf.String() + } stderrDone := make(chan error, 1) go func() { - errScanner := bufio.NewScanner(stderr) - for errScanner.Scan() { - errBuf.WriteString(errScanner.Text()) - errBuf.WriteByte('\n') - } - stderrDone <- errScanner.Err() + _, err := io.Copy(&errBuf, stderr) + stderrDone <- 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(), - }) + var ( + outputTokens int + readErr error + ) + switch profile.OutputFormat { + case "stream-json": + outputTokens, readErr = emitStreamJSONLines(ctx, stdout, sink, spec.RunID, &outBuf) + case "codex-json": + outputTokens, readErr = emitCodexJSONLines(ctx, stdout, sink, spec.RunID, &outBuf) + case "claude-json": + outputTokens, readErr = emitClaudeJSONLines(ctx, stdout, sink, spec.RunID, &outBuf) + default: + outputTokens, readErr = emitStdoutChunks(ctx, stdout, sink, spec.RunID, &outBuf) } - - if err := scanner.Err(); err != nil { + if readErr != nil { _ = <-stderrDone _ = cmd.Wait() _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, - Error: fmt.Sprintf("read stdout: %v", err), + Error: fmt.Sprintf("read stdout: %v", readErr), Timestamp: time.Now(), }) - return fmt.Errorf("cli adapter: read stdout: %w", err) + return combinedOutput(), fmt.Errorf("cli adapter: read stdout: %w", readErr) } waitErr := cmd.Wait() stderrErr := <-stderrDone if waitErr != nil { + if ctx.Err() != nil { + _ = sink.Emit(context.Background(), runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeCancelled, + Message: "cli execution cancelled", + Timestamp: time.Now(), + }) + return combinedOutput(), runtime.ErrRunCancelled + } _ = 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) + return combinedOutput(), fmt.Errorf("cli adapter: command exited with error: %w", waitErr) } if stderrErr != nil && !errors.Is(stderrErr, os.ErrClosed) { _ = sink.Emit(ctx, runtime.RuntimeEvent{ @@ -96,10 +113,10 @@ func (c *CLI) executeOneShot(ctx context.Context, spec runtime.ExecutionSpec, pr Error: fmt.Sprintf("read stderr: %v", stderrErr), Timestamp: time.Now(), }) - return fmt.Errorf("cli adapter: read stderr: %w", stderrErr) + return combinedOutput(), fmt.Errorf("cli adapter: read stderr: %w", stderrErr) } - return sink.Emit(ctx, runtime.RuntimeEvent{ + return combinedOutput(), sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeComplete, Message: "cli execution complete", @@ -111,6 +128,216 @@ func (c *CLI) executeOneShot(ctx context.Context, spec runtime.ExecutionSpec, pr }) } +func emitStdoutChunks(ctx context.Context, stdout io.Reader, sink runtime.EventSink, runID string, outBuf *strings.Builder) (int, error) { + buf := make([]byte, 4096) + outputTokens := 0 + + for { + n, err := stdout.Read(buf) + if n > 0 { + delta := string(buf[:n]) + outBuf.WriteString(delta) + outputTokens += len(strings.Fields(delta)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeDelta, + Delta: delta, + Timestamp: time.Now(), + }) + } + if errors.Is(err, io.EOF) { + return outputTokens, nil + } + if err != nil { + return outputTokens, err + } + } +} + +// emitStreamJSONLines reads JSON Lines from stdout (one JSON object per line) +// and emits assistant content (delta=true) as RuntimeEvent deltas. Other +// event types are written to outBuf so they remain available for diagnostics +// but are not pushed to the sink. +func emitStreamJSONLines(ctx context.Context, stdout io.Reader, sink runtime.EventSink, runID string, outBuf *strings.Builder) (int, error) { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + outputTokens := 0 + + for scanner.Scan() { + line := scanner.Bytes() + outBuf.Write(line) + outBuf.WriteByte('\n') + + trimmed := strings.TrimSpace(string(line)) + if trimmed == "" || trimmed[0] != '{' { + continue + } + var ev struct { + Type string `json:"type"` + Role string `json:"role"` + Content string `json:"content"` + Delta bool `json:"delta"` + Error string `json:"error"` + } + if err := json.Unmarshal(line, &ev); err != nil { + continue + } + switch ev.Type { + case "message": + if ev.Role != "assistant" || ev.Content == "" { + continue + } + outputTokens += len(strings.Fields(ev.Content)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeDelta, + Delta: ev.Content, + Timestamp: time.Now(), + }) + case "error": + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: ev.Error, + Timestamp: time.Now(), + }) + } + } + if err := scanner.Err(); err != nil { + return outputTokens, err + } + return outputTokens, nil +} + +// emitClaudeJSONLines parses `claude -p --output-format stream-json +// --include-partial-messages` JSONL output. Token-level text arrives as +// stream_event with event.type == "content_block_delta" and +// event.delta.type == "text_delta". Other event types (system, assistant, +// result, rate_limit_event, ...) are written to outBuf for diagnostics but +// not pushed to the sink to avoid duplicate text. +func emitClaudeJSONLines(ctx context.Context, stdout io.Reader, sink runtime.EventSink, runID string, outBuf *strings.Builder) (int, error) { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64*1024), 8*1024*1024) + outputTokens := 0 + + for scanner.Scan() { + line := scanner.Bytes() + outBuf.Write(line) + outBuf.WriteByte('\n') + + trimmed := strings.TrimSpace(string(line)) + if trimmed == "" || trimmed[0] != '{' { + continue + } + var ev struct { + Type string `json:"type"` + Event struct { + Type string `json:"type"` + Delta struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"delta"` + } `json:"event"` + IsError bool `json:"is_error"` + Result string `json:"result"` + } + if err := json.Unmarshal(line, &ev); err != nil { + continue + } + switch ev.Type { + case "stream_event": + if ev.Event.Type != "content_block_delta" || ev.Event.Delta.Type != "text_delta" || ev.Event.Delta.Text == "" { + continue + } + outputTokens += len(strings.Fields(ev.Event.Delta.Text)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeDelta, + Delta: ev.Event.Delta.Text, + Timestamp: time.Now(), + }) + case "result": + if ev.IsError && ev.Result != "" { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: ev.Result, + Timestamp: time.Now(), + }) + } + } + } + if err := scanner.Err(); err != nil { + return outputTokens, err + } + return outputTokens, nil +} + +// emitCodexJSONLines parses `codex exec --json` JSONL output. Codex emits +// one event per line; assistant text arrives in a single `item.completed` +// event with item.type == "agent_message" rather than token-by-token. +func emitCodexJSONLines(ctx context.Context, stdout io.Reader, sink runtime.EventSink, runID string, outBuf *strings.Builder) (int, error) { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + outputTokens := 0 + + for scanner.Scan() { + line := scanner.Bytes() + outBuf.Write(line) + outBuf.WriteByte('\n') + + trimmed := strings.TrimSpace(string(line)) + if trimmed == "" || trimmed[0] != '{' { + continue + } + var ev struct { + Type string `json:"type"` + Item struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"item"` + Message string `json:"message"` + Error struct { + Message string `json:"message"` + } `json:"error"` + } + if err := json.Unmarshal(line, &ev); err != nil { + continue + } + switch ev.Type { + case "item.completed": + if ev.Item.Type != "agent_message" || ev.Item.Text == "" { + continue + } + outputTokens += len(strings.Fields(ev.Item.Text)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeDelta, + Delta: ev.Item.Text, + Timestamp: time.Now(), + }) + case "error": + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: ev.Message, + Timestamp: time.Now(), + }) + case "turn.failed": + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: ev.Error.Message, + Timestamp: time.Now(), + }) + } + } + if err := scanner.Err(); err != nil { + return outputTokens, err + } + return outputTokens, nil +} + func extractPrompt(input map[string]any) string { if input == nil { return "" diff --git a/apps/node/internal/adapters/cli/oneshot/cli_test.go b/apps/node/internal/adapters/cli/oneshot/cli_test.go index c7e5652..f4fb365 100644 --- a/apps/node/internal/adapters/cli/oneshot/cli_test.go +++ b/apps/node/internal/adapters/cli/oneshot/cli_test.go @@ -57,6 +57,117 @@ func TestCLIExecuteOneShotPassesPromptAsArg(t *testing.T) { } } +func TestCLIExecuteOneShotStreamJSONParsesAssistantContent(t *testing.T) { + testutil.RequireUnixShell(t) + + script := `cat <<'EOF' +{"type":"init","timestamp":"2026-05-03T00:00:00.000Z","session_id":"abc"} +{"type":"message","role":"user","content":"hi"} +{"type":"message","role":"assistant","content":"Hello","delta":true} +{"type":"message","role":"assistant","content":" world","delta":true} +{"type":"result","status":"success"} +EOF` + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "streamy": { + Command: "sh", + Args: []string{"-c", script, "sh"}, + OutputFormat: "stream-json", + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + if err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-sj", + Model: "streamy", + Input: map[string]any{"prompt": "hi"}, + }, sink); err != nil { + t.Fatalf("execute: %v", err) + } + combined := testutil.CollectDeltas(sink.Events()) + if combined != "Hello world" { + t.Fatalf("expected stream-json deltas to concat to %q, got %q", "Hello world", combined) + } +} + +func TestCLIExecuteOneShotCodexJSONParsesAgentMessage(t *testing.T) { + testutil.RequireUnixShell(t) + + script := `cat <<'EOF' +{"type":"thread.started","thread_id":"019d-aaa"} +{"type":"turn.started"} +{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"Hi there."}} +{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":3}} +EOF` + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "codexy": { + Command: "sh", + Args: []string{"-c", script, "sh"}, + OutputFormat: "codex-json", + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + if err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-cj", + Model: "codexy", + Input: map[string]any{"prompt": "hi"}, + }, sink); err != nil { + t.Fatalf("execute: %v", err) + } + combined := testutil.CollectDeltas(sink.Events()) + if combined != "Hi there." { + t.Fatalf("expected codex-json deltas to be %q, got %q", "Hi there.", combined) + } +} + +func TestCLIExecuteOneShotClaudeJSONParsesContentBlockDeltas(t *testing.T) { + testutil.RequireUnixShell(t) + + script := `cat <<'EOF' +{"type":"system","subtype":"init","session_id":"abc"} +{"type":"stream_event","event":{"type":"message_start","message":{"id":"msg_1"}}} +{"type":"stream_event","event":{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" world"}}} +{"type":"assistant","message":{"id":"msg_1","content":[{"type":"text","text":"Hello world"}]}} +{"type":"stream_event","event":{"type":"content_block_stop","index":0}} +{"type":"stream_event","event":{"type":"message_stop"}} +{"type":"result","subtype":"success","is_error":false,"result":"Hello world"} +EOF` + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "claudish": { + Command: "sh", + Args: []string{"-c", script, "sh"}, + OutputFormat: "claude-json", + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + if err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-cl", + Model: "claudish", + Input: map[string]any{"prompt": "hi"}, + }, sink); err != nil { + t.Fatalf("execute: %v", err) + } + combined := testutil.CollectDeltas(sink.Events()) + if combined != "Hello world" { + t.Fatalf("expected claude-json deltas to be %q, got %q", "Hello world", combined) + } +} + func TestCLIExecuteOneShotDrainsStderrConcurrently(t *testing.T) { testutil.RequireUnixShell(t) @@ -92,3 +203,45 @@ func TestCLIExecuteOneShotDrainsStderrConcurrently(t *testing.T) { t.Fatalf("expected reply:hello in deltas, got %q", combined) } } + +func TestCLIExecuteOneShotStreamsStdoutChunks(t *testing.T) { + testutil.RequireUnixShell(t) + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "chunky": { + Command: "sh", + Args: []string{ + "-c", + `printf "first"; sleep 0.2; printf "second"`, + "sh", + }, + }, + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-chunky", + Model: "chunky", + Input: map[string]any{"prompt": "unused"}, + }, sink) + if err != nil { + t.Fatalf("execute: %v", err) + } + + var deltas []string + for _, event := range sink.Events() { + if event.Type == noderuntime.EventTypeDelta { + deltas = append(deltas, event.Delta) + } + } + if len(deltas) < 2 { + t.Fatalf("expected stdout chunks to be emitted separately, got %v", deltas) + } + if got := strings.Join(deltas, ""); got != "firstsecond" { + t.Fatalf("combined deltas: got %q want firstsecond", got) + } +} diff --git a/apps/node/internal/adapters/cli/persistent/codex_exec_test.go b/apps/node/internal/adapters/cli/persistent/codex_exec_test.go new file mode 100644 index 0000000..b73e8ba --- /dev/null +++ b/apps/node/internal/adapters/cli/persistent/codex_exec_test.go @@ -0,0 +1,225 @@ +package persistent_test + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "go.uber.org/zap" + + clipkg "iop/apps/node/internal/adapters/cli" + "iop/apps/node/internal/adapters/cli/internal/testutil" + noderuntime "iop/apps/node/internal/runtime" + "iop/packages/config" +) + +func TestCLIStartSkipsCodexExecPersistentStartup(t *testing.T) { + testutil.RequireUnixShell(t) + + codex := writeFakeCodex(t, `#!/usr/bin/env sh +echo "codex should not be started during adapter startup" >&2 +exit 42 +`) + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "codex": { + Command: codex, + Args: []string{"exec", "--dangerously-bypass-approvals-and-sandbox"}, + Persistent: true, + }, + }, + } + + c := clipkg.New(cfg, zap.NewNop()) + if err := c.Start(context.Background()); err != nil { + t.Fatalf("Start should not invoke codex exec: %v", err) + } +} + +func TestCLIExecuteCodexExecPersistentResumesLogicalSession(t *testing.T) { + testutil.RequireUnixShell(t) + + codex := writeFakeCodex(t, `#!/usr/bin/env sh +if [ "$1" != "exec" ]; then + echo "unexpected command: $*" >&2 + exit 2 +fi + +if [ "$2" = "resume" ]; then + shift 2 + for arg in "$@"; do + if [ "$arg" = "--color" ]; then + echo "resume received unsupported --color option" >&2 + exit 12 + fi + done + + session="" + prompt="" + for arg in "$@"; do + case "$arg" in + --*) continue ;; + esac + if [ -z "$session" ]; then + session="$arg" + else + prompt="$arg" + fi + done + + printf "session id: %s\n" "$session" >&2 + printf "reply:resume:%s:%s\n" "$session" "$prompt" + exit 0 +fi + +last="" +for arg in "$@"; do + last="$arg" +done +printf "session id: codex-session-1\n" >&2 +printf "reply:first:%s\n" "$last" +`) + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "codex": { + Command: codex, + Args: []string{ + "exec", + "--dangerously-bypass-approvals-and-sandbox", + "--color", + "never", + "--skip-git-repo-check", + }, + Persistent: true, + }, + }, + } + + c := clipkg.New(cfg, zap.NewNop()) + ctx := context.Background() + + first := &testutil.FakeSink{} + if err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-1", + Model: "codex", + SessionID: "session-a", + Input: map[string]any{"prompt": "first"}, + }, first); err != nil { + t.Fatalf("first execute: %v", err) + } + if combined := testutil.CollectDeltas(first.Events()); !strings.Contains(combined, "reply:first:first") { + t.Fatalf("first execute output: %q", combined) + } + + second := &testutil.FakeSink{} + if err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-2", + Model: "codex", + SessionID: "session-a", + Input: map[string]any{"prompt": "second"}, + }, second); err != nil { + t.Fatalf("second execute: %v", err) + } + if combined := testutil.CollectDeltas(second.Events()); !strings.Contains(combined, "reply:resume:codex-session-1:second") { + t.Fatalf("second execute output: %q", combined) + } +} + +func TestCLIExecuteCodexExecJSONFormatStreamsAgentMessageAndResumes(t *testing.T) { + testutil.RequireUnixShell(t) + + codex := writeFakeCodex(t, `#!/usr/bin/env sh +if [ "$1" != "exec" ]; then + echo "unexpected command: $*" >&2 + exit 2 +fi + +if [ "$2" = "resume" ]; then + shift 2 + session="" + prompt="" + for arg in "$@"; do + case "$arg" in + --*) continue ;; + esac + if [ -z "$session" ]; then + session="$arg" + else + prompt="$arg" + fi + done + printf '{"type":"thread.started","thread_id":"%s"}\n' "$session" + printf '{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"resume:%s:%s"}}\n' "$session" "$prompt" + exit 0 +fi + +last="" +for arg in "$@"; do + last="$arg" +done +printf '{"type":"thread.started","thread_id":"codex-session-json-1"}\n' +printf '{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"first:%s"}}\n' "$last" +`) + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "codex": { + Command: codex, + Args: []string{ + "exec", + "--dangerously-bypass-approvals-and-sandbox", + "--color", + "never", + "--skip-git-repo-check", + "--json", + }, + Persistent: true, + OutputFormat: "codex-json", + }, + }, + } + + c := clipkg.New(cfg, zap.NewNop()) + ctx := context.Background() + + first := &testutil.FakeSink{} + if err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-1", + Model: "codex", + SessionID: "session-a", + Input: map[string]any{"prompt": "hello"}, + }, first); err != nil { + t.Fatalf("first execute: %v", err) + } + if combined := testutil.CollectDeltas(first.Events()); combined != "first:hello" { + t.Fatalf("first deltas: got %q want %q", combined, "first:hello") + } + + second := &testutil.FakeSink{} + if err := c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-2", + Model: "codex", + SessionID: "session-a", + Input: map[string]any{"prompt": "again"}, + }, second); err != nil { + t.Fatalf("second execute: %v", err) + } + if combined := testutil.CollectDeltas(second.Events()); combined != "resume:codex-session-json-1:again" { + t.Fatalf("second deltas: got %q", combined) + } +} + +func writeFakeCodex(t *testing.T, script string) string { + t.Helper() + + dir := t.TempDir() + path := filepath.Join(dir, "codex") + if err := os.WriteFile(path, []byte(script), 0o755); err != nil { + t.Fatalf("write fake codex: %v", err) + } + return path +} diff --git a/configs/edge.yaml b/configs/edge.yaml index 9e8579b..2d5a0e2 100644 --- a/configs/edge.yaml +++ b/configs/edge.yaml @@ -13,7 +13,7 @@ metrics: console: adapter: "cli" - model: "codex" + model: "claude" session_id: "default" background: false timeout_sec: 120 @@ -36,18 +36,26 @@ nodes: args: - "-p" - "--dangerously-skip-permissions" + - "--output-format" + - "stream-json" + - "--include-partial-messages" + - "--verbose" env: [] persistent: false terminal: false + output_format: "claude-json" gemini: command: "gemini" args: - - "-p" - "--approval-mode" - "yolo" + - "--output-format" + - "stream-json" + - "-p" env: [] persistent: false terminal: false + output_format: "stream-json" codex: command: "codex" args: @@ -56,11 +64,13 @@ nodes: - "--color" - "never" - "--skip-git-repo-check" + - "--json" env: [] - # persistent: true enables logical worker session reuse. - # Set to false to use one-shot mode (new process per request). + # codex exec is non-interactive; persistent mode uses codex exec + # resume per logical session instead of keeping stdin open. persistent: true terminal: false + output_format: "codex-json" runtime: concurrency: 4 workspace_root: "/tmp/iop/workspace" diff --git a/packages/config/config.go b/packages/config/config.go index 7b9653e..b726e94 100644 --- a/packages/config/config.go +++ b/packages/config/config.go @@ -103,6 +103,11 @@ type CLIProfileConf struct { Terminal bool `mapstructure:"terminal" yaml:"terminal"` ResponseIdleTimeoutMS int `mapstructure:"response_idle_timeout_ms" yaml:"response_idle_timeout_ms"` StartupIdleTimeoutMS int `mapstructure:"startup_idle_timeout_ms" yaml:"startup_idle_timeout_ms"` + // OutputFormat declares how the adapter should parse the CLI stdout. + // Empty (default) means raw stdout chunks are forwarded as deltas. + // "stream-json" means each stdout line is a JSON event and assistant + // content with delta=true is forwarded as the delta payload. + OutputFormat string `mapstructure:"output_format" yaml:"output_format"` } func Load(cfgFile string) (*NodeConfig, error) { @@ -145,7 +150,7 @@ func setEdgeDefaults(v *viper.Viper) { v.SetDefault("metrics.port", 9092) v.SetDefault("tls.enabled", false) v.SetDefault("console.adapter", "cli") - v.SetDefault("console.model", "codex") + v.SetDefault("console.model", "claude") v.SetDefault("console.session_id", "default") v.SetDefault("console.background", false) v.SetDefault("console.timeout_sec", 120) diff --git a/packages/config/config_test.go b/packages/config/config_test.go index 9dfa58a..c496475 100644 --- a/packages/config/config_test.go +++ b/packages/config/config_test.go @@ -25,8 +25,8 @@ func TestLoadEdge_ConsoleTimeoutDefault(t *testing.T) { 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) + if cfg.Console.Model != "claude" { + t.Fatalf("expected default console.model=%q, got %q", "claude", cfg.Console.Model) } }