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
This commit is contained in:
parent
96f79fcd08
commit
157a8a7076
13 changed files with 1051 additions and 74 deletions
|
|
@ -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 <prompt>`, `codex exec --dangerously-bypass-approvals-and-sandbox`처럼 headless+bypass 조합으로 설정한다. Claude는 `claude -p --dangerously-skip-permissions --output-format stream-json --include-partial-messages --verbose <prompt>` 형태로 호출되어 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 ... <prompt>` 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] <Codex 응답 텍스트>
|
||||
|
|
|
|||
|
|
@ -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] <empty>\n", alias)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(out, "[node-%s-message] %s\n", alias, message)
|
||||
}
|
||||
|
||||
func normalizeConsoleSessionID(id string) string {
|
||||
if id == "" {
|
||||
return "default"
|
||||
|
|
|
|||
156
apps/edge/cmd/edge/console_events.go
Normal file
156
apps/edge/cmd/edge/console_events.go
Normal file
|
|
@ -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<empty>\n", s.prefix)
|
||||
return
|
||||
}
|
||||
if !s.endedWithNewline {
|
||||
fmt.Fprintln(s.out)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *consoleResponseStream) FinishIfStarted() {
|
||||
if s.started {
|
||||
s.Finish()
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 <prompt>` 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] <Codex 응답 텍스트>
|
||||
[node-message] <Claude 응답 텍스트 (토큰 단위로 점진적 표시)>
|
||||
```
|
||||
|
||||
## 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 <prompt>`, `codex exec --dangerously-bypass-approvals-and-sandbox`.
|
||||
|
||||
## Transport
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
120
apps/node/internal/adapters/cli/codex_exec.go
Normal file
120
apps/node/internal/adapters/cli/codex_exec.go
Normal file
|
|
@ -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 ""
|
||||
}
|
||||
|
|
@ -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 ""
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
225
apps/node/internal/adapters/cli/persistent/codex_exec_test.go
Normal file
225
apps/node/internal/adapters/cli/persistent/codex_exec_test.go
Normal file
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue