diff --git a/.clinerules b/.clinerules index e3c77d1..07fbe60 100644 --- a/.clinerules +++ b/.clinerules @@ -5,12 +5,12 @@ - 요청 범위를 넘는 변경을 하지 않는다. - 불확실하면 단정하지 말고 후보를 제시한다. -아래 성격의 요청의 경우, 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 읽고 수행한다. 자동으로 수행하지 않도록한다. +`agent-ops/rules/project/rules.md`와 `agent-ops/rules/private/rules.md`는 세션 최초 1회 읽는다. 파일이 없으면 무시한다. +`agent-ops/rules/project/domain/**` 경로의 domain rule은 domain 정보를 읽어야할때, 해당 domain 최초 진입 시 1회 읽고, 이미 읽은 rule은 반복하지 않는다. + +아래 성격의 요청은 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 작업 최초 1회 읽고 수행한다. 자동으로 수행하지 않는다. - agent-ops 초기화 - domain rule 생성 - skill 생성 - git commit / push - agent-ops 업데이트 / 진입 파일 재적용 - -`agent-ops/rules/project/rules.md`와 -`agent-ops/rules/private/rules.md`를 읽고 작업을 시작한다. 파일이 없을경우 무시한다. diff --git a/.cursorrules b/.cursorrules index 6648b75..07fbe60 100644 --- a/.cursorrules +++ b/.cursorrules @@ -5,12 +5,12 @@ - 요청 범위를 넘는 변경을 하지 않는다. - 불확실하면 단정하지 말고 후보를 제시한다. -아래 요청은 `agent-ops/skills/common/router.md`를 읽고 수행한다. +`agent-ops/rules/project/rules.md`와 `agent-ops/rules/private/rules.md`는 세션 최초 1회 읽는다. 파일이 없으면 무시한다. +`agent-ops/rules/project/domain/**` 경로의 domain rule은 domain 정보를 읽어야할때, 해당 domain 최초 진입 시 1회 읽고, 이미 읽은 rule은 반복하지 않는다. + +아래 성격의 요청은 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 작업 최초 1회 읽고 수행한다. 자동으로 수행하지 않는다. - agent-ops 초기화 - domain rule 생성 - skill 생성 -- git commit / git push +- git commit / push - agent-ops 업데이트 / 진입 파일 재적용 - -`agent-ops/rules/project/rules.md`와 -`agent-ops/rules/private/rules.md`를 읽고 작업을 시작한다. 파일이 없을경우 무시한다. diff --git a/AGENTS.md b/AGENTS.md index 6648b75..07fbe60 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,12 +5,12 @@ - 요청 범위를 넘는 변경을 하지 않는다. - 불확실하면 단정하지 말고 후보를 제시한다. -아래 요청은 `agent-ops/skills/common/router.md`를 읽고 수행한다. +`agent-ops/rules/project/rules.md`와 `agent-ops/rules/private/rules.md`는 세션 최초 1회 읽는다. 파일이 없으면 무시한다. +`agent-ops/rules/project/domain/**` 경로의 domain rule은 domain 정보를 읽어야할때, 해당 domain 최초 진입 시 1회 읽고, 이미 읽은 rule은 반복하지 않는다. + +아래 성격의 요청은 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 작업 최초 1회 읽고 수행한다. 자동으로 수행하지 않는다. - agent-ops 초기화 - domain rule 생성 - skill 생성 -- git commit / git push +- git commit / push - agent-ops 업데이트 / 진입 파일 재적용 - -`agent-ops/rules/project/rules.md`와 -`agent-ops/rules/private/rules.md`를 읽고 작업을 시작한다. 파일이 없을경우 무시한다. diff --git a/CLAUDE.md b/CLAUDE.md index 6648b75..07fbe60 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -5,12 +5,12 @@ - 요청 범위를 넘는 변경을 하지 않는다. - 불확실하면 단정하지 말고 후보를 제시한다. -아래 요청은 `agent-ops/skills/common/router.md`를 읽고 수행한다. +`agent-ops/rules/project/rules.md`와 `agent-ops/rules/private/rules.md`는 세션 최초 1회 읽는다. 파일이 없으면 무시한다. +`agent-ops/rules/project/domain/**` 경로의 domain rule은 domain 정보를 읽어야할때, 해당 domain 최초 진입 시 1회 읽고, 이미 읽은 rule은 반복하지 않는다. + +아래 성격의 요청은 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 작업 최초 1회 읽고 수행한다. 자동으로 수행하지 않는다. - agent-ops 초기화 - domain rule 생성 - skill 생성 -- git commit / git push +- git commit / push - agent-ops 업데이트 / 진입 파일 재적용 - -`agent-ops/rules/project/rules.md`와 -`agent-ops/rules/private/rules.md`를 읽고 작업을 시작한다. 파일이 없을경우 무시한다. diff --git a/GEMINI.md b/GEMINI.md index 6648b75..07fbe60 100644 --- a/GEMINI.md +++ b/GEMINI.md @@ -5,12 +5,12 @@ - 요청 범위를 넘는 변경을 하지 않는다. - 불확실하면 단정하지 말고 후보를 제시한다. -아래 요청은 `agent-ops/skills/common/router.md`를 읽고 수행한다. +`agent-ops/rules/project/rules.md`와 `agent-ops/rules/private/rules.md`는 세션 최초 1회 읽는다. 파일이 없으면 무시한다. +`agent-ops/rules/project/domain/**` 경로의 domain rule은 domain 정보를 읽어야할때, 해당 domain 최초 진입 시 1회 읽고, 이미 읽은 rule은 반복하지 않는다. + +아래 성격의 요청은 사용자가 명시적으로 요청한 경우에만 `agent-ops/skills/common/router.md`를 작업 최초 1회 읽고 수행한다. 자동으로 수행하지 않는다. - agent-ops 초기화 - domain rule 생성 - skill 생성 -- git commit / git push +- git commit / push - agent-ops 업데이트 / 진입 파일 재적용 - -`agent-ops/rules/project/rules.md`와 -`agent-ops/rules/private/rules.md`를 읽고 작업을 시작한다. 파일이 없을경우 무시한다. diff --git a/agent-task/opencode_sse_stream/code_review_cloud_G07_0.log b/agent-task/opencode_sse_stream/code_review_cloud_G07_0.log new file mode 100644 index 0000000..78dac66 --- /dev/null +++ b/agent-task/opencode_sse_stream/code_review_cloud_G07_0.log @@ -0,0 +1,204 @@ + + +# Code Review Reference - OPENCODE_SSE + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every section below is filled in. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-05-11 +task=opencode_sse_stream, plan=0, tag=OPENCODE_SSE + +## 이 파일을 읽는 리뷰 에이전트에게 + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료 후 반드시 아래 순서로 아카이브하세요. + +1. `CODE_REVIEW-cloud-G07.md` → `code_review_cloud_G07_N.log` (N = 기존 code_review_*.log 수) +2. `PLAN-cloud-G07.md` → `plan_cloud_G07_M.log` (M = 기존 plan_*.log 수) +3. PASS인 경우 `complete.log` 작성 후 종료. WARN/FAIL인 경우 새 routed plan + review 스텁 작성. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [OPENCODE_SSE-1] CLI adapter에 opencode SSE mode 추가 | [x] | +| [OPENCODE_SSE-2] opencode profile 예시와 문서를 SSE mode로 정리 | [x] | + +## 계획 대비 변경 사항 + +- 계획은 `executeOpencodeSSE`가 동기 scanner loop으로 SSE를 처리하는 구조를 암시했으나, 구현은 SSE 이벤트를 read하는 goroutine + `select { case <-ctx.Done(); case e := <-events }` 형태로 분리했다. 이유: `ctx.Done()` 즉시 응답성을 보장하면서도 abort 호출/cancelled emit/`ErrRunCancelled` 반환을 깔끔히 직렬화하기 위함이다. 외부 계약(`runtime.Adapter`/`RuntimeEvent`)은 그대로다. +- 계획의 `permission.asked` 응답에서 `--dangerously-skip-permissions`가 없으면 `reject`로 답하도록 했고, 응답 호출은 background context + 5초 timeout으로 묶었다. SSE drain 도중 context 취소 시 reply가 ctx와 함께 죽지 않도록 하기 위함이다. +- `opencodeAbort` 또한 cancel finalize 경로에서 별도 `context.WithTimeout(2s)`로 호출한다. 원래 ctx는 이미 done 상태이므로 abort 요청 자체가 즉시 실패하지 않게 하기 위해서다. +- 검증 명령은 plan에 명시된 그대로 사용했다. 대체 없음. + +## 주요 설계 결정 + +- `opencodeSSESession`은 `serverURL`, `sessionID`, `cmd`, `owned` 4개 필드만 보관한다. `owned=true`일 때만 local server process를 kill한다. `--attach`로 외부 server를 사용할 때는 process 소유권이 없으므로 cache 제거만 한다. +- HTTP 라우팅 키: SSE 이벤트가 `properties.sessionID`를 포함할 때만 본 세션과 일치 여부를 검사하고, 누락된 경우는 통과시킨다. 이렇게 해서 서버측 이벤트 스키마 변화에 대한 호환성 여유를 둔다. +- complete usage는 `session.next.step.ended`의 `tokens.input/output`에서 갱신하고 0이 아닌 값만 덮어쓴다. complete event는 `session.idle` 또는 `session.status.status=="idle"` 중 먼저 들어오는 쪽에서 emit한다. +- `parseOpencodeRunArgs`는 `opencode run` 호환 옵션만 인식하고 알 수 없는 토큰은 무시한다. legacy `--format json` 등은 silent ignore 되어 사용자 설정 마이그레이션 시 잡음을 줄인다. +- `closeOpencodeSession`은 attach 세션에 대해 process kill을 시도하지 않는다. abort 요청은 cancel 경로에서만 호출한다 — TerminateSession은 cache drop만 한다 (외부 server 정책에 맡긴다). + +## 리뷰어를 위한 체크포인트 + +- `runtime.Adapter`, `runtime.EventSink`, protobuf `RunEvent` 계약이 변경되지 않았는지 확인한다. +- `opencode-sse`는 `profile.Mode` 내부 분기로만 동작하고 기존 one-shot/persistent/codex-exec 경로를 깨지 않는지 확인한다. +- SSE tests는 `httptest.Server`를 사용하고 실제 opencode binary나 실제 모델에 의존하지 않는지 확인한다. +- `session.next.text.delta`가 즉시 `RuntimeEvent` delta로 emit되는지, complete usage는 `session.next.step.ended`에서 반영되는지 확인한다. +- context cancel이 opencode abort 호출, cancelled event, `runtime.ErrRunCancelled`를 모두 만족하는지 확인한다. +- config/docs 변경이 `mode: "opencode-sse"` 방향을 설명하고 legacy `opencode-json` stdout 경로와 혼동되지 않는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. + +### OPENCODE_SSE-1 중간 검증 +``` +$ go test -count=1 ./apps/node/internal/adapters/cli +ok iop/apps/node/internal/adapters/cli 87.356s +``` + +### OPENCODE_SSE-2 중간 검증 +``` +$ go test -count=1 ./packages/config ./apps/edge/internal/node ./apps/node/internal/adapters +ok iop/packages/config 0.006s +ok iop/apps/edge/internal/node 0.003s +ok iop/apps/node/internal/adapters 0.004s +``` + +### 최종 검증 +``` +$ go test -count=1 ./apps/node/internal/adapters/cli ./apps/node/internal/adapters ./apps/edge/internal/node ./apps/edge/internal/transport ./packages/config +ok iop/apps/node/internal/adapters/cli 87.406s +ok iop/apps/node/internal/adapters 0.005s +ok iop/apps/edge/internal/node 0.004s +ok iop/apps/edge/internal/transport 0.007s +ok iop/packages/config 0.006s +``` + +``` +$ go test -count=1 ./... +? iop/apps/control-plane/cmd/control-plane [no test files] +ok iop/apps/edge/cmd/edge 0.006s +? iop/apps/edge/internal/bootstrap [no test files] +ok iop/apps/edge/internal/node 0.006s +ok iop/apps/edge/internal/transport 0.009s +? iop/apps/node/cmd/node [no test files] +ok iop/apps/node/internal/adapters 0.006s +ok iop/apps/node/internal/adapters/cli 87.369s +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status 0.004s +? iop/apps/node/internal/adapters/mock [no test files] +? iop/apps/node/internal/adapters/ollama [no test files] +? iop/apps/node/internal/adapters/vllm [no test files] +ok iop/apps/node/internal/bootstrap 0.162s +ok iop/apps/node/internal/node 0.009s +ok iop/apps/node/internal/router 0.004s +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store 0.049s +ok iop/apps/node/internal/transport 0.006s +? iop/apps/worker/cmd/worker [no test files] +? iop/packages/auth [no test files] +ok iop/packages/config 0.015s +? iop/packages/jobs [no test files] +? iop/packages/metadata [no test files] +? iop/packages/observability [no test files] +? iop/packages/policy [no test files] +? iop/packages/version [no test files] +? iop/proto/gen/iop [no test files] +``` + +``` +$ rg --sort path -n 'opencode-sse|opencode-json|opencode run --format json|session.next.text.delta' apps/node apps/edge configs packages agent-task/opencode_sse_stream +apps/node/README.md:95:`opencode`는 profile `mode: "opencode-sse"`로 설정해 `opencode serve`의 HTTP/SSE 인터페이스를 사용한다. `profile.Command`가 가리키는 opencode 바이너리를 `serve --hostname 127.0.0.1 --port 0`으로 띄우고 `/event` SSE에서 `session.next.text.delta`를 즉시 `RuntimeEvent` delta로 relay한다. 외부에서 이미 실행 중인 server에 연결하려면 `--attach `을 args에 추가한다. legacy `output_format: "opencode-json"` stdout JSONL 경로는 `opencode run --format json` 사용 시에만 의미가 있다. +apps/node/internal/adapters/cli/cli.go:29: modeOpencodeSSE = "opencode-sse" +apps/node/internal/adapters/cli/cli_internal_test.go:320: {opencodeJSONEmitter{}, "opencode-json"}, +apps/node/internal/adapters/cli/cli_internal_test.go:331: expectedKeys := []string{"stream-json", "claude-json", "codex-json", "opencode-json", "cline-json"} +apps/node/internal/adapters/cli/emitters.go:33: "opencode-json": {emitter: opencodeJSONEmitter{}, scanBufMax: 4 * 1024 * 1024}, +apps/node/internal/adapters/cli/emitters.go:282:// --- opencode-json emitter --- +apps/node/internal/adapters/cli/emitters.go:286:func (opencodeJSONEmitter) Name() string { return "opencode-json" } +apps/node/internal/adapters/cli/oneshot_blackbox_test.go:185: OutputFormat: "opencode-json", +apps/node/internal/adapters/cli/oneshot_blackbox_test.go:201: t.Fatalf("expected opencode-json deltas to be %q, got %q", "Hi from opencode.", combined) +apps/node/internal/adapters/cli/oneshot_blackbox_test.go:218: OutputFormat: "opencode-json", +apps/node/internal/adapters/cli/oneshot_blackbox_test.go:260: OutputFormat: "opencode-json", +apps/node/internal/adapters/cli/oneshot_blackbox_test.go:300: OutputFormat: "opencode-json", +apps/node/internal/adapters/cli/opencode_sse.go:370: case "session.next.text.delta": +apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go:130: Mode: "opencode-sse", +apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go:165: "type": "session.next.text.delta", +apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go:172: "type": "session.next.text.delta", +apps/node/internal/adapters/factory_internal_test.go:169: Mode: "opencode-sse", +apps/node/internal/adapters/factory_internal_test.go:178: if prof.Mode != "opencode-sse" { +apps/node/internal/adapters/factory_internal_test.go:192: "mode": "opencode-sse", +apps/node/internal/adapters/factory_internal_test.go:204: if prof.Mode != "opencode-sse" { +apps/edge/README.md:18:같은 `cli` 어댑터 안에 `claude`, `gemini`, `codex`, `opencode`, `cline` profile을 포함할 수 있다. 예를 들어 `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`, `cline -y --json --config /config/.cline/profiles/ollama-dgx `처럼 headless 조합으로 설정한다. `opencode`는 `mode: "opencode-sse"` profile로 `opencode serve`의 HTTP/SSE 인터페이스를 사용하며, args의 `--model`, `--dangerously-skip-permissions`, `--title` 등은 그대로 유지하면 된다. +apps/edge/internal/node/mapper_test.go:152: Mode: "opencode-sse", +apps/edge/internal/node/mapper_test.go:178: if p.Mode != "opencode-sse" { +apps/edge/internal/node/mapper_test.go:179: t.Errorf("mode: got %q, want %q", p.Mode, "opencode-sse") +configs/edge.yaml:96: mode: "opencode-sse" +packages/config/config_test.go:429: mode: "opencode-sse" +packages/config/config_test.go:445: if prof.Mode != "opencode-sse" { +agent-task/opencode_sse_stream/CODE_REVIEW-cloud-G07.md:43:- `opencode-sse`는 `profile.Mode` 내부 분기로만 동작하고 기존 one-shot/persistent/codex-exec 경로를 깨지 않는지 확인한다. +agent-task/opencode_sse_stream/CODE_REVIEW-cloud-G07.md:45:- `session.next.text.delta`가 즉시 `RuntimeEvent` delta로 emit되는지, complete usage는 `session.next.step.ended`에서 반영되는지 확인한다. +agent-task/opencode_sse_stream/CODE_REVIEW-cloud-G07.md:47:- config/docs 변경이 `mode: "opencode-sse"` 방향을 설명하고 legacy `opencode-json` stdout 경로와 혼동되지 않는지 확인한다. +agent-task/opencode_sse_stream/CODE_REVIEW-cloud-G07.md:82:$ rg --sort path -n 'opencode-sse|opencode-json|opencode run --format json|session.next.text.delta' apps/node apps/edge configs packages agent-task/opencode_sse_stream +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:11:`opencode run --format json`은 IOP의 stdout JSONL 파이프라인에 연결되어 있지만, OpenCode CLI 구현상 완성된 text part 중심으로 출력되어 토큰 단위 delta stream을 안정적으로 얻기 어렵다. OpenCode server는 `/event` SSE와 `/session/{id}/prompt_async`를 제공하며, 이 경로에는 `session.next.text.delta` 이벤트가 있다. IOP 공통 인터페이스는 `runtime.Adapter.Execute(ctx, spec, sink)`와 `RuntimeEvent`이므로, opencode 내부 실행 방식만 SSE client로 바꾸면 Edge/Node transport 계약은 유지된다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:45:- `opencode-json` stdout JSONL parsing은 `apps/node/internal/adapters/cli/oneshot_blackbox_test.go`와 `cli_internal_test.go`에 있다. SSE 기반 `session.next.text.delta` relay는 테스트가 없다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:46:- `CLI.Execute`의 mode 분기는 `codex-exec`, `persistent`, one-shot만 검증되어 있다. `opencode-sse` 분기, session reuse, `REQUIRE_EXISTING` 실패는 테스트가 없다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:48:- edge YAML의 `mode` 로딩은 `codex-exec` 예시만 있다. `opencode-sse` profile 예시는 없다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:120:`executeOneShot`은 prompt를 subprocess argv 뒤에 붙이는 구조라 `opencode run --format json`의 stdout event만 읽을 수 있다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:142: modeOpencodeSSE = "opencode-sse" +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:165:`Execute`는 `opencode-sse`를 persistent/one-shot보다 먼저 처리한다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:185:- SSE `session.next.text.delta`는 `RuntimeEvent{Type: delta, Delta: properties.delta}`로 즉시 emit한다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:205: - `TestCLIExecuteOpencodeSSE_StreamsTextDeltas`: fake `/event`, `/session`, `/prompt_async`로 `session.next.text.delta` 2개와 idle을 보내고 delta concat, complete, usage를 검증한다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:243: output_format: "opencode-json" +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:267:`configs/edge.yaml`의 opencode profile을 `mode: "opencode-sse"`로 바꾸고 `args`는 prompt 전송 옵션만 남긴다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:282: mode: "opencode-sse" +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:285:`apps/node/README.md`와 `apps/edge/README.md`는 opencode 예시를 `opencode serve` + SSE mode로 설명한다. stdout JSONL `opencode-json` emitter는 legacy/run mode로 남긴다. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:289:- [ ] `configs/edge.yaml`: opencode profile `mode: "opencode-sse"` 적용, `run`/`--format json` 제거. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:290:- [ ] `apps/node/README.md`: CLI adapter 설명에 opencode는 `opencode-sse` mode에서 server SSE를 사용한다고 기록. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:300: - `TestLoadEdge_OpencodeSSEProfile`: YAML에서 `mode: "opencode-sse"`, `--model`, `--dangerously-skip-permissions`가 로드되는지 확인. +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:344:rg --sort path -n 'opencode-sse|opencode-json|opencode run --format json|session.next.text.delta' apps/node apps/edge configs packages agent-task/opencode_sse_stream +agent-task/opencode_sse_stream/PLAN-cloud-G07.md:347:예상 결과: `opencode-sse`는 새 mode/문서/테스트에 나타난다. `opencode-json`과 `opencode run --format json`은 legacy stdout JSONL 경로로 의도적으로 남은 항목만 나타난다. +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every section: completion table, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. + +## 코드리뷰 결과 + +### 종합 판정 + +FAIL + +### 차원별 평가 + +| 차원 | 평가 | +|------|------| +| correctness | Fail | +| completeness | Fail | +| test coverage | Fail | +| API contract | Fail | +| code quality | Pass | +| plan deviation | Pass | +| verification trust | Warn | + +### 발견된 문제 + +- Required: `apps/node/internal/adapters/cli/opencode_sse.go:562`에서 `payload["model"] = opts.Model`로 문자열을 보내고 있다. OpenCode `1.14.46`의 `/session/{sessionID}/prompt_async` schema는 `model`을 `{providerID, modelID}` 객체로 요구한다. 현재 `configs/edge.yaml`의 `--model ollama-dgx/qwen3.6:35b-a3b-bf16` 조합은 실서버에서 400 응답을 낼 수 있다. `--model provider/model`을 첫 `/` 기준으로 분리해 `map[string]any{"providerID": provider, "modelID": model}`로 보내고, fake server test가 prompt body shape를 검증하도록 수정한다. +- Required: `apps/node/internal/adapters/cli/opencode_sse.go:401`의 `session.status` 처리에서 `status`를 문자열 `"idle"`로만 읽는다. OpenCode schema의 `status`는 `SessionStatus` 객체이며 idle은 보통 `{"type":"idle"}` 형태다. 서버가 `session.status`로 완료 상태를 알리면 현재 adapter는 이를 무시하고 stream close나 timeout까지 대기할 수 있다. `status` 문자열과 객체를 모두 해석하는 helper를 추가하고 `session.status` 객체만으로 complete 되는 regression test를 추가한다. + +### 다음 단계 + +FAIL: Required 문제를 해결하는 새 routed plan/review 파일을 작성하고 루프를 계속한다. diff --git a/agent-task/opencode_sse_stream/code_review_cloud_G07_1.log b/agent-task/opencode_sse_stream/code_review_cloud_G07_1.log new file mode 100644 index 0000000..0625371 --- /dev/null +++ b/agent-task/opencode_sse_stream/code_review_cloud_G07_1.log @@ -0,0 +1,135 @@ + + +# Code Review Reference - REVIEW_OPENCODE_SSE + +## 개요 + +date=2026-05-11 +task=opencode_sse_stream, plan=1, tag=REVIEW_OPENCODE_SSE + +## 이 파일을 읽는 리뷰 에이전트에게 + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료 후 반드시 아래 순서로 아카이브하세요. + +1. `CODE_REVIEW-cloud-G07.md` → `code_review_cloud_G07_N.log` (N = 기존 code_review_*.log 수) +2. `PLAN-cloud-G07.md` → `plan_cloud_G07_M.log` (M = 기존 plan_*.log 수) +3. PASS인 경우 `complete.log` 작성 후 종료. WARN/FAIL인 경우 새 routed plan + review 스텁 작성. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_OPENCODE_SSE-1] prompt_async model payload를 OpenCode schema에 맞춘다 | [x] | +| [REVIEW_OPENCODE_SSE-2] session.status 객체 idle을 complete로 처리한다 | [x] | + +## 계획 대비 변경 사항 + +- 계획의 helper 이름을 그대로 채용했다(`opencodeModelPayload`, `opencodeStatusIdle`). complete event emit 중복 제거는 plan이 허용한 작은 refactor로 한정해 `emitOpencodeComplete` helper를 추가하고 `session.idle`, `session.status` (object/string), scan-EOF 3개 경로에서 공유한다. +- 1차 구현의 `StreamsTextDeltas` 테스트가 `--model m1`을 사용했는데, model 객체화 후에는 invalid 값이 되어 동일 케이스를 `prov/m1`로 정정하고 body assertion(`model.providerID`, `model.modelID`, `parts[0].text`)을 추가했다. 별도 케이스 추가 대신 기존 테스트를 강화한 것은 plan이 "TestCLIExecuteOpencodeSSE_StreamsTextDeltas에 prompt body assertion을 추가한다"고 명시했기 때문이다. +- model parse 실패는 prompt_async POST 이전에 검출되므로 fake server는 prompt 요청을 받지 않는다. 그 흐름을 검증하는 `TestCLIExecuteOpencodeSSE_InvalidModelEmitsError`를 추가했다. error event emit은 `executeOpencodeSSE`의 prompt POST 실패 분기에서 처리한다. +- 검증 명령은 plan에 명시된 그대로 사용했다. 대체 없음. + +## 주요 설계 결정 + +- `opencodeModelPayload`는 첫 `/`만으로 split한다. 즉 `openrouter/google/gemini-2.5-pro`는 provider=`openrouter`, modelID=`google/gemini-2.5-pro`로 들어간다. OpenRouter 같은 multi-segment model id 호환을 우선했다. +- `opencodeStatusIdle`은 `string`과 `map[string]any{"type":"idle"}` 둘 다 idle로 인정한다. 옛 OpenCode build와 신 build 양쪽 호환을 위해 두 형태를 모두 처리한다. +- complete emit은 단일 helper로 통일했지만, cancel 경로(`finalize`)와 error 경로는 별도 emit/return을 유지한다. complete 의미와 cancel/error 의미를 섞지 않기 위함이다. + +## 리뷰어를 위한 체크포인트 + +- `--model provider/model`이 `prompt_async` body에서 `{"providerID": provider, "modelID": model}` 객체로 전송되는지 확인한다. +- model string이 잘못된 경우 silent ignore가 아니라 error event와 error return으로 드러나는지 확인한다. +- `session.status`의 `status`가 `{"type":"idle"}` 객체일 때 complete event가 emit되는지 확인한다. +- 새 tests가 fake server에서 request body와 status object event를 실제로 검증하는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. + +### REVIEW_OPENCODE_SSE-1 중간 검증 +``` +$ go test -count=1 ./apps/node/internal/adapters/cli +ok iop/apps/node/internal/adapters/cli 87.559s +``` + +### REVIEW_OPENCODE_SSE-2 중간 검증 +``` +$ go test -count=1 ./apps/node/internal/adapters/cli +ok iop/apps/node/internal/adapters/cli 87.559s +``` + +### 최종 검증 +``` +$ go test -count=1 ./apps/node/internal/adapters/cli ./apps/node/internal/adapters ./apps/edge/internal/node ./apps/edge/internal/transport ./packages/config +ok iop/apps/node/internal/adapters/cli 87.535s +ok iop/apps/node/internal/adapters 0.004s +ok iop/apps/edge/internal/node 0.004s +ok iop/apps/edge/internal/transport 0.006s +ok iop/packages/config 0.007s +``` + +``` +$ go test -count=1 ./... +? iop/apps/control-plane/cmd/control-plane [no test files] +ok iop/apps/edge/cmd/edge 0.005s +? iop/apps/edge/internal/bootstrap [no test files] +ok iop/apps/edge/internal/node 0.007s +ok iop/apps/edge/internal/transport 0.007s +? iop/apps/node/cmd/node [no test files] +ok iop/apps/node/internal/adapters 0.005s +ok iop/apps/node/internal/adapters/cli 87.567s +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status 0.003s +? iop/apps/node/internal/adapters/mock [no test files] +? iop/apps/node/internal/adapters/ollama [no test files] +? iop/apps/node/internal/adapters/vllm [no test files] +ok iop/apps/node/internal/bootstrap 0.165s +ok iop/apps/node/internal/node 0.009s +ok iop/apps/node/internal/router 0.004s +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store 0.023s +ok iop/apps/node/internal/transport 0.006s +? iop/apps/worker/cmd/worker [no test files] +? iop/packages/auth [no test files] +ok iop/packages/config 0.009s +? iop/packages/jobs [no test files] +? iop/packages/metadata [no test files] +? iop/packages/observability [no test files] +? iop/packages/policy [no test files] +? iop/packages/version [no test files] +? iop/proto/gen/iop [no test files] +``` + +## 코드리뷰 결과 + +### 종합 판정 + +PASS + +### 차원별 평가 + +| 차원 | 평가 | +|------|------| +| correctness | Pass | +| completeness | Pass | +| test coverage | Pass | +| API contract | Pass | +| code quality | Pass | +| plan deviation | Pass | +| verification trust | Pass | + +### 발견된 문제 + +없음 + +### 다음 단계 + +PASS: active plan/review를 archive하고 `complete.log`를 작성해 종료한다. diff --git a/agent-task/opencode_sse_stream/complete.log b/agent-task/opencode_sse_stream/complete.log new file mode 100644 index 0000000..c9af2ee --- /dev/null +++ b/agent-task/opencode_sse_stream/complete.log @@ -0,0 +1,22 @@ +# Task Complete - opencode_sse_stream + +완료 일시: 2026-05-11T13:33:38Z + +## 요약 + +OpenCode adapter의 SSE 기반 stream 실행을 구현하고 API 계약 보정까지 2회 리뷰 루프로 완료했다. + +## 루프 이력 + +| Loop | Plan | Review | Verdict | +|------|------|--------|---------| +| 0 | `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | FAIL | +| 1 | `plan_cloud_G07_1.log` | `code_review_cloud_G07_1.log` | PASS | + +## 최종 리뷰 요약 + +- `opencode-sse` mode가 OpenCode server `/event` SSE와 `/session/{id}/prompt_async`를 통해 delta stream을 `RuntimeEvent`로 relay한다. +- `prompt_async`의 `model` payload를 OpenCode schema에 맞는 `{providerID, modelID}` 객체로 전송한다. +- `session.idle`과 `session.status` 객체 idle 모두 complete event로 처리한다. +- fake OpenCode server 기반 blackbox tests와 helper unit tests가 model payload, status idle, error/cancel/session 흐름을 검증한다. +- 대상 패키지 검증과 repository-wide `go test -count=1 ./...`가 통과했다. diff --git a/agent-task/opencode_sse_stream/plan_cloud_G07_0.log b/agent-task/opencode_sse_stream/plan_cloud_G07_0.log new file mode 100644 index 0000000..fd6d85b --- /dev/null +++ b/agent-task/opencode_sse_stream/plan_cloud_G07_0.log @@ -0,0 +1,349 @@ + + +# Opencode SSE Stream Plan + +## 이 파일을 읽는 구현 에이전트에게 + +**반드시 마지막에 `CODE_REVIEW-cloud-G07.md`의 모든 섹션을 실제 구현 내용과 검증 출력으로 채워야 한다. 이 파일 작성 전에는 작업이 완료된 것이 아니다.** 아래 체크리스트를 끝까지 수행하고, 중간/최종 검증 명령의 실제 stdout/stderr를 review 파일에 붙인다. review 파일의 `이 파일을 읽는 리뷰 에이전트에게` 섹션에 있는 archive 지시(`*.log` rename, `complete.log` 작성)는 구현 에이전트가 실행하지 않는다. + +## 배경 + +`opencode run --format json`은 IOP의 stdout JSONL 파이프라인에 연결되어 있지만, OpenCode CLI 구현상 완성된 text part 중심으로 출력되어 토큰 단위 delta stream을 안정적으로 얻기 어렵다. OpenCode server는 `/event` SSE와 `/session/{id}/prompt_async`를 제공하며, 이 경로에는 `session.next.text.delta` 이벤트가 있다. IOP 공통 인터페이스는 `runtime.Adapter.Execute(ctx, spec, sink)`와 `RuntimeEvent`이므로, opencode 내부 실행 방식만 SSE client로 바꾸면 Edge/Node transport 계약은 유지된다. + +## 분석 결과 + +### 읽은 파일 + +- `agent-ops/rules/project/rules.md` +- `agent-ops/rules/project/domain/node/rules.md` +- `agent-ops/rules/project/domain/edge/rules.md` +- `agent-ops/rules/project/domain/platform-common/rules.md` +- `agent-ops/skills/common/router.md` +- `agent-ops/skills/common/plan/SKILL.md` +- `go.mod` +- `apps/node/internal/adapters/cli/cli.go` +- `apps/node/internal/adapters/cli/oneshot.go` +- `apps/node/internal/adapters/cli/emitters.go` +- `apps/node/internal/adapters/cli/codex_exec.go` +- `apps/node/internal/adapters/cli/persistent.go` +- `apps/node/internal/adapters/factory.go` +- `apps/edge/internal/node/mapper.go` +- `packages/config/config.go` +- `proto/iop/runtime.proto` +- `configs/edge.yaml` +- `apps/node/README.md` +- `apps/edge/README.md` +- `apps/node/internal/adapters/cli/oneshot_blackbox_test.go` +- `apps/node/internal/adapters/cli/cli_internal_test.go` +- `apps/node/internal/adapters/factory_internal_test.go` +- `apps/edge/internal/node/mapper_test.go` +- `apps/edge/internal/transport/server_test.go` +- `packages/config/config_test.go` + +### 테스트 커버리지 공백 + +- `opencode-json` stdout JSONL parsing은 `apps/node/internal/adapters/cli/oneshot_blackbox_test.go`와 `cli_internal_test.go`에 있다. SSE 기반 `session.next.text.delta` relay는 테스트가 없다. +- `CLI.Execute`의 mode 분기는 `codex-exec`, `persistent`, one-shot만 검증되어 있다. `opencode-sse` 분기, session reuse, `REQUIRE_EXISTING` 실패는 테스트가 없다. +- cancel 시 opencode `/session/{id}/abort` 호출과 IOP cancelled event 변환은 테스트가 없다. +- edge YAML의 `mode` 로딩은 `codex-exec` 예시만 있다. `opencode-sse` profile 예시는 없다. + +### 심볼 참조 + +- renamed/removed symbols: none. +- 새 symbol 후보: `modeOpencodeSSE`, `opencodeSSESession`, `executeOpencodeSSE`, `parseOpencodeRunArgs`. 새 symbol이므로 기존 call site는 없다. + +### 범위 결정 근거 + +- `runtime.Adapter`, `runtime.EventSink`, `RunEvent`, Edge-Node TCP/protobuf transport는 변경하지 않는다. 공통 인터페이스가 이미 stream relay 계약을 제공한다. +- `proto/iop/runtime.proto`는 변경하지 않는다. `CLIProfileConfig.mode`, `args`, `env`가 이미 edge에서 node로 전달된다. +- OpenCode 실제 binary나 실제 모델을 호출하는 통합 테스트는 제외한다. `httptest.Server`로 SSE/HTTP 계약을 검증하면 결정적이다. +- `ollama`, `vllm`, `mock`, `edge console` 실행 로직은 제외한다. opencode profile 내부 실행 방식만 바꾼다. + +### 빌드 등급 + +- build=`cloud-G07`, review=`cloud-G07`: HTTP/SSE protocol parsing, session lifecycle, cancellation, cross-package config examples가 포함되어 local-only 검증보다 높은 리뷰가 필요하다. + +## 의존 관계 및 구현 순서 + +1. `OPENCODE_SSE-1`: CLI adapter 내부 mode/session/SSE 실행기 구현. +2. `OPENCODE_SSE-2`: 설정 예시와 문서 갱신. +3. 각 항목 중간 검증 후 최종 전체 테스트. + +### [OPENCODE_SSE-1] CLI adapter에 opencode SSE mode 추가 + +#### 문제 + +`CLI`는 persistent session map과 codex exec session map만 가진다. + +```go +// apps/node/internal/adapters/cli/cli.go:55 +type CLI struct { + mu sync.Mutex + profiles map[string]config.CLIProfileConf + sessions map[sessionKey]*profileSession + codexSessions map[sessionKey]*codexExecSession + logger *zap.Logger +``` + +`Start`는 `codex-exec`만 persistent startup에서 제외한다. + +```go +// apps/node/internal/adapters/cli/cli.go:93 +names := make([]string, 0, len(c.profiles)) +for name := range c.profiles { + if c.profiles[name].Persistent && c.profiles[name].Mode != "codex-exec" { + names = append(names, name) + } +} +``` + +`Execute`는 `codex-exec`, persistent, one-shot만 분기한다. + +```go +// apps/node/internal/adapters/cli/cli.go:153 +func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { + agent := cliAgentName(spec) + profile, ok := c.profiles[agent] + if !ok { + return fmt.Errorf("cli adapter: unknown agent %q", agent) + } + if profile.Mode == "codex-exec" { + return c.executeCodexExec(ctx, spec, profile, sink) + } + if profile.Persistent { + return c.executePersistent(ctx, spec, profile, sink) + } + return c.executeOneShot(ctx, spec, profile, sink) +} +``` + +`executeOneShot`은 prompt를 subprocess argv 뒤에 붙이는 구조라 `opencode run --format json`의 stdout event만 읽을 수 있다. + +```go +// apps/node/internal/adapters/cli/oneshot.go:17 +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 +} +``` + +#### 해결 방법 + +`apps/node/internal/adapters/cli/opencode_sse.go`를 새로 추가한다. mode별 파일을 둔 `codex_exec.go` 패턴을 따른다. + +`cli.go`의 session state와 dispatch를 다음 형태로 확장한다. + +```go +// after +const ( + modeCodexExec = "codex-exec" + modeOpencodeSSE = "opencode-sse" +) + +type opencodeSSESession struct { + key sessionKey + serverURL string + sessionID string + cmd *exec.Cmd + owned bool + mu sync.Mutex +} + +type CLI struct { + mu sync.Mutex + profiles map[string]config.CLIProfileConf + sessions map[sessionKey]*profileSession + codexSessions map[sessionKey]*codexExecSession + opencodeSessions map[sessionKey]*opencodeSSESession + logger *zap.Logger + // ... +} +``` + +`Execute`는 `opencode-sse`를 persistent/one-shot보다 먼저 처리한다. + +```go +// after +if profile.Mode == modeCodexExec { + return c.executeCodexExec(ctx, spec, profile, sink) +} +if profile.Mode == modeOpencodeSSE { + return c.executeOpencodeSSE(ctx, spec, profile, sink) +} +``` + +`executeOpencodeSSE`는 다음 계약을 지킨다. + +- `profile.Args`는 `opencode run` 호환 옵션으로 해석한다: `--attach`, `--model`, `--agent`, `--variant`, `--title`, `--dangerously-skip-permissions`, `--dir`. +- `--attach `이 있으면 외부 server를 사용하고 process를 띄우지 않는다. +- `--attach`가 없으면 `profile.Command serve --hostname 127.0.0.1 --port 0`을 실행하고 stdout/stderr에서 `opencode server listening on http://...`를 파싱한다. +- `SessionModeRequireExisting`이고 `(target, session_id)`에 cached opencode session이 없으면 codex/persistent와 같은 형태의 error를 반환한다. +- `GET /event` SSE를 먼저 열고, session이 없으면 `POST /session`으로 생성한다. +- `POST /session/{id}/prompt_async` body는 `{parts:[{type:"text",text:prompt}], model?, agent?, variant?}`로 보낸다. +- SSE `session.next.text.delta`는 `RuntimeEvent{Type: delta, Delta: properties.delta}`로 즉시 emit한다. +- SSE `session.next.step.ended`의 `tokens.input/output`은 complete event usage에 보관한다. +- SSE `session.status` idle 또는 `session.idle`에서 `RuntimeEvent{Type: complete, Message:"opencode sse execution complete"}`를 emit하고 종료한다. +- SSE `session.error`는 error event emit 후 error return한다. +- `permission.asked`는 `--dangerously-skip-permissions`가 있으면 `/permission/{id}/reply` with `{"reply":"once"}`, 없으면 `{"reply":"reject"}`로 답한다. +- `ctx.Done()`이면 `/session/{id}/abort`를 best-effort로 호출하고 cancelled event를 emit한 뒤 `runtime.ErrRunCancelled`를 반환한다. + +`Stop`과 `TerminateSession`은 `opencodeSessions`도 닫는다. owned local server process는 kill하고, attach session은 process kill 없이 cache만 제거하며 가능하면 abort를 호출한다. + +#### 수정 파일 및 체크리스트 + +- [ ] `apps/node/internal/adapters/cli/cli.go`: mode constants, `opencodeSessions` map, `Start` skip, `Execute` branch, `Stop`, `TerminateSession` 확장. +- [ ] `apps/node/internal/adapters/cli/opencode_sse.go`: HTTP/SSE client, option parser, local server startup, session resolve, prompt send, event loop, abort/close helpers 구현. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go`: httptest 기반 execute/cancel/error/session tests 추가. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_internal_test.go`: args parser, SSE data parser unit tests 추가. + +#### 테스트 작성 + +- 작성한다. +- `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go` + - `TestCLIExecuteOpencodeSSE_StreamsTextDeltas`: fake `/event`, `/session`, `/prompt_async`로 `session.next.text.delta` 2개와 idle을 보내고 delta concat, complete, usage를 검증한다. + - `TestCLIExecuteOpencodeSSE_RequireExistingWithoutSessionErrors`: cached session이 없고 `SessionModeRequireExisting`이면 error. + - `TestCLIExecuteOpencodeSSE_SessionErrorEmitsRuntimeError`: `session.error`를 error event와 error return으로 검증한다. + - `TestCLIExecuteOpencodeSSE_ContextCancelAbortsSession`: context cancel 시 `/session/{id}/abort` 호출, cancelled event, `runtime.ErrRunCancelled`. +- `apps/node/internal/adapters/cli/opencode_sse_internal_test.go` + - `TestParseOpencodeRunArgs`: `--attach`, `--model provider/model`, `--agent`, `--variant`, `--dangerously-skip-permissions` parsing. + - `TestDecodeSSEDataLine`: `data: {...}` parsing과 non-data skip. + +#### 중간 검증 + +```bash +go test -count=1 ./apps/node/internal/adapters/cli +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +### [OPENCODE_SSE-2] opencode profile 예시와 문서를 SSE mode로 정리 + +#### 문제 + +예시 config는 opencode를 `run --format json` stdout JSONL mode로 둔다. + +```yaml +# configs/edge.yaml:85 + opencode: + command: "/config/.npm-global/bin/opencode" + args: + - "run" + - "--title" + - "untitle" + - "--model" + - "ollama-dgx/qwen3.6:35b-a3b-bf16" + - "--format" + - "json" + - "--dangerously-skip-permissions" + env: [] + persistent: false + terminal: false + output_format: "opencode-json" +``` + +`CLIProfileConf`에는 이미 mode string이 있어 새 config field 없이 opencode 전용 구현을 선택할 수 있다. + +```go +// packages/config/config.go:124 +type CLIProfileConf struct { + Command string `mapstructure:"command" yaml:"command"` + Args []string `mapstructure:"args" yaml:"args"` + Env []string `mapstructure:"env" yaml:"env"` + Persistent bool `mapstructure:"persistent" yaml:"persistent"` + 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 string `mapstructure:"output_format" yaml:"output_format"` + CompletionMarker CompletionMarkerConf `mapstructure:"completion_marker" yaml:"completion_marker"` + Mode string `mapstructure:"mode" yaml:"mode"` + ResumeArgs []string `mapstructure:"resume_args" yaml:"resume_args"` +} +``` + +#### 해결 방법 + +`configs/edge.yaml`의 opencode profile을 `mode: "opencode-sse"`로 바꾸고 `args`는 prompt 전송 옵션만 남긴다. + +```yaml +# after + opencode: + command: "/config/.npm-global/bin/opencode" + args: + - "--title" + - "untitle" + - "--model" + - "ollama-dgx/qwen3.6:35b-a3b-bf16" + - "--dangerously-skip-permissions" + env: [] + persistent: false + terminal: false + mode: "opencode-sse" +``` + +`apps/node/README.md`와 `apps/edge/README.md`는 opencode 예시를 `opencode serve` + SSE mode로 설명한다. stdout JSONL `opencode-json` emitter는 legacy/run mode로 남긴다. + +#### 수정 파일 및 체크리스트 + +- [ ] `configs/edge.yaml`: opencode profile `mode: "opencode-sse"` 적용, `run`/`--format json` 제거. +- [ ] `apps/node/README.md`: CLI adapter 설명에 opencode는 `opencode-sse` mode에서 server SSE를 사용한다고 기록. +- [ ] `apps/edge/README.md`: 수동 테스트 예시의 opencode 명령 설명 갱신. +- [ ] `packages/config/config_test.go`: `TestLoadEdge_OpencodeSSEProfile` 추가. +- [ ] `apps/edge/internal/node/mapper_test.go`: `TestBuildConfigPayload_CLIProfiles` 또는 새 테스트에서 `Mode` 전달 검증. +- [ ] `apps/node/internal/adapters/factory_internal_test.go`: proto/legacy settings에서 `mode`가 node CLI config로 들어오는지 검증. + +#### 테스트 작성 + +- 작성한다. +- `packages/config/config_test.go` + - `TestLoadEdge_OpencodeSSEProfile`: YAML에서 `mode: "opencode-sse"`, `--model`, `--dangerously-skip-permissions`가 로드되는지 확인. +- `apps/edge/internal/node/mapper_test.go` + - `Mode`와 `ResumeArgs`가 `CLIProfileConfig`로 전달되는지 확인. +- `apps/node/internal/adapters/factory_internal_test.go` + - typed proto와 legacy struct fallback 모두 `mode` 값을 유지하는지 확인. + +#### 중간 검증 + +```bash +go test -count=1 ./packages/config ./apps/edge/internal/node ./apps/node/internal/adapters +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/node/internal/adapters/cli/cli.go` | OPENCODE_SSE-1 | +| `apps/node/internal/adapters/cli/opencode_sse.go` | OPENCODE_SSE-1 | +| `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go` | OPENCODE_SSE-1 | +| `apps/node/internal/adapters/cli/opencode_sse_internal_test.go` | OPENCODE_SSE-1 | +| `configs/edge.yaml` | OPENCODE_SSE-2 | +| `apps/node/README.md` | OPENCODE_SSE-2 | +| `apps/edge/README.md` | OPENCODE_SSE-2 | +| `packages/config/config_test.go` | OPENCODE_SSE-2 | +| `apps/edge/internal/node/mapper_test.go` | OPENCODE_SSE-2 | +| `apps/node/internal/adapters/factory_internal_test.go` | OPENCODE_SSE-2 | + +## 최종 검증 + +```bash +go test -count=1 ./apps/node/internal/adapters/cli ./apps/node/internal/adapters ./apps/edge/internal/node ./apps/edge/internal/transport ./packages/config +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +```bash +go test -count=1 ./... +``` + +예상 결과: repository-wide pass. Go test cache는 허용하지 않는다. + +```bash +rg --sort path -n 'opencode-sse|opencode-json|opencode run --format json|session.next.text.delta' apps/node apps/edge configs packages agent-task/opencode_sse_stream +``` + +예상 결과: `opencode-sse`는 새 mode/문서/테스트에 나타난다. `opencode-json`과 `opencode run --format json`은 legacy stdout JSONL 경로로 의도적으로 남은 항목만 나타난다. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 전체 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/opencode_sse_stream/plan_cloud_G07_1.log b/agent-task/opencode_sse_stream/plan_cloud_G07_1.log new file mode 100644 index 0000000..8edec81 --- /dev/null +++ b/agent-task/opencode_sse_stream/plan_cloud_G07_1.log @@ -0,0 +1,160 @@ + + +# Review Fix Plan - OpenCode SSE API Contract + +## 이 파일을 읽는 구현 에이전트에게 + +**반드시 마지막에 `CODE_REVIEW-cloud-G07.md`의 모든 섹션을 실제 구현 내용과 검증 출력으로 채워야 한다. 이 파일 작성 전에는 작업이 완료된 것이 아니다.** 아래 체크리스트를 끝까지 수행하고, 중간/최종 검증 명령의 실제 stdout/stderr를 review 파일에 붙인다. review 파일의 `이 파일을 읽는 리뷰 에이전트에게` 섹션에 있는 archive 지시(`*.log` rename, `complete.log` 작성)는 구현 에이전트가 실행하지 않는다. + +## 배경 + +1차 구현은 테스트를 통과했지만 OpenCode server OpenAPI 계약과 맞지 않는 부분이 확인됐다. `/session/{sessionID}/prompt_async`의 `model` payload는 객체여야 하고, `session.status.status`는 문자열이 아니라 `SessionStatus` 객체다. 이 두 문제는 실서버에서 prompt 전송 실패나 stream completion 누락으로 이어질 수 있다. + +## 분석 결과 + +### 읽은 파일 + +- `agent-task/opencode_sse_stream/plan_cloud_G07_0.log` +- `agent-task/opencode_sse_stream/code_review_cloud_G07_0.log` +- `apps/node/internal/adapters/cli/opencode_sse.go` +- `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go` +- `apps/node/internal/adapters/cli/opencode_sse_internal_test.go` +- `configs/edge.yaml` +- local OpenCode `1.14.46` `/doc` OpenAPI response + +### 테스트 커버리지 공백 + +- `--model provider/model`이 `/prompt_async` body에서 `{providerID, modelID}` 객체로 전송되는지 검증하지 않는다. +- `session.status` 이벤트가 `{"status":{"type":"idle"}}` 형태로 올 때 complete 되는지 검증하지 않는다. + +### 심볼 참조 + +- renamed/removed symbols: none. +- 새 helper 후보: `opencodeModelPayload`, `opencodeIdleStatus`. + +### 범위 결정 근거 + +- `runtime.Adapter`, protobuf, Edge transport는 변경하지 않는다. 문제는 opencode SSE adapter 내부 HTTP payload/event parsing 계약에 한정된다. +- `configs/edge.yaml`의 opencode profile 이름과 args는 유지한다. +- 실제 opencode binary/model 통합 테스트는 제외한다. `httptest.Server`와 local OpenAPI 계약에 맞춘 unit/blackbox tests로 검증한다. + +### 빌드 등급 + +- build=`cloud-G07`, review=`cloud-G07`: HTTP API 계약과 streaming completion 조건을 수정하므로 기존 등급을 유지한다. + +### [REVIEW_OPENCODE_SSE-1] prompt_async model payload를 OpenCode schema에 맞춘다 + +#### 문제 + +```go +// apps/node/internal/adapters/cli/opencode_sse.go:562 +if opts.Model != "" { + payload["model"] = opts.Model +} +``` + +OpenCode `/session/{sessionID}/prompt_async` schema는 `model`을 `{providerID, modelID}` 객체로 요구한다. 현재 구현은 문자열을 보내므로 `--model ollama-dgx/qwen3.6:35b-a3b-bf16` 사용 시 실서버가 Bad Request를 반환할 수 있다. + +#### 해결 방법 + +`--model provider/model`을 첫 `/` 기준으로 분리한다. provider 또는 model이 비어 있거나 `/`가 없으면 명확한 error를 반환하고 `RuntimeEvent error`를 emit한다. + +```go +// after +if opts.Model != "" { + model, err := opencodeModelPayload(opts.Model) + if err != nil { + return err + } + payload["model"] = model +} +``` + +#### 수정 파일 및 체크리스트 + +- [ ] `apps/node/internal/adapters/cli/opencode_sse.go`: `opencodeModelPayload` helper 추가, `opencodePromptAsync`에서 객체 payload 사용. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_internal_test.go`: 정상 split, `/` 없는 model, provider 빈 값, model 빈 값 테스트. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go`: fake server의 `lastPromptBody`를 JSON decode해서 `model.providerID`, `model.modelID`, `parts[0].text`를 검증. + +#### 테스트 작성 + +- 작성한다. +- `TestOpencodeModelPayload` 또는 동등한 이름으로 model parsing boundary를 검증한다. +- `TestCLIExecuteOpencodeSSE_StreamsTextDeltas`에 prompt body assertion을 추가한다. + +#### 중간 검증 + +```bash +go test -count=1 ./apps/node/internal/adapters/cli +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +### [REVIEW_OPENCODE_SSE-2] session.status 객체 idle을 complete로 처리한다 + +#### 문제 + +```go +// apps/node/internal/adapters/cli/opencode_sse.go:401 +case "session.status": + if status, ok := stringFromProp(e.ev.Properties, "status"); ok && status == "idle" { +``` + +OpenCode `EventSessionStatus.properties.status`는 `SessionStatus` 객체다. `{"status":{"type":"idle"}}` 이벤트가 오면 현재 구현은 complete를 emit하지 않는다. + +#### 해결 방법 + +문자열 `"idle"`과 객체 `{"type":"idle"}`를 모두 처리하는 helper를 추가한다. + +```go +// after +case "session.status": + if opencodeStatusIdle(e.ev.Properties) { + return emitOpencodeComplete(ctx, spec.RunID, inputTokens, outputTokens, sink) + } +``` + +complete event 중복 코드는 작게 helper로 묶어도 된다. 단, 이 refactor는 `opencode_sse.go` 안에 한정한다. + +#### 수정 파일 및 체크리스트 + +- [ ] `apps/node/internal/adapters/cli/opencode_sse.go`: `opencodeStatusIdle` helper 추가, `session.status` branch 수정. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_internal_test.go`: string idle, object idle, busy/retry/non-map false 테스트. +- [ ] `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go`: `session.idle` 없이 `session.status` object idle만 보내도 run이 complete 되는 regression test 추가. + +#### 테스트 작성 + +- 작성한다. +- `TestCLIExecuteOpencodeSSE_SessionStatusObjectIdleCompletes`를 추가한다. + +#### 중간 검증 + +```bash +go test -count=1 ./apps/node/internal/adapters/cli +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/node/internal/adapters/cli/opencode_sse.go` | REVIEW_OPENCODE_SSE-1, REVIEW_OPENCODE_SSE-2 | +| `apps/node/internal/adapters/cli/opencode_sse_internal_test.go` | REVIEW_OPENCODE_SSE-1, REVIEW_OPENCODE_SSE-2 | +| `apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go` | REVIEW_OPENCODE_SSE-1, REVIEW_OPENCODE_SSE-2 | + +## 최종 검증 + +```bash +go test -count=1 ./apps/node/internal/adapters/cli ./apps/node/internal/adapters ./apps/edge/internal/node ./apps/edge/internal/transport ./packages/config +``` + +예상 결과: package pass. Go test cache는 허용하지 않는다. + +```bash +go test -count=1 ./... +``` + +예상 결과: repository-wide pass. Go test cache는 허용하지 않는다. + +모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 전체 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/apps/edge/README.md b/apps/edge/README.md index c2495f9..b5237bc 100644 --- a/apps/edge/README.md +++ b/apps/edge/README.md @@ -1,42 +1,43 @@ -# edge — Edge API Gateway +# edge — Edge Execution Group Controller -외부 클라이언트에 OpenAI-compatible HTTP API를 노출하고, -IOP 내부 TCP 프로토콜을 통해 node로 요청을 라우팅한다. +여러 Node를 하나의 로컬 실행 그룹으로 묶고, IOP 내부 TCP/protobuf 프로토콜을 통해 adapter execution 요청과 이벤트 스트림을 중계한다. + +Edge는 현재 단계에서 OpenAI-compatible HTTP API gateway가 아니라 Node registry, node configuration, runtime routing, stream relay를 검증하는 백엔드 실행 그룹 컨트롤러다. 외부 HTTP 호환 계층은 node/transport 경계가 안정화된 뒤 별도 표면으로 추가한다. **현재 상태: 초기 구현** node 등록/레지스트리/transport와 edge 콘솔 기반 수동 통신 테스트가 구현되어 있다. -## 원격 Codex CLI 수동 테스트 +## 원격 CLI adapter 수동 테스트 `bin/edge.sh`는 edge 서버를 열고 입력 프롬프트를 제공한다. -`bin/node.sh`는 원격 edge 주소로 접속해 등록한 뒤, edge에서 보낸 `RunRequest`를 cli/codex one-shot process로 실행한다. +`bin/node.sh`는 원격 edge 주소로 접속해 등록한 뒤, edge에서 보낸 `RunRequest`를 `adapter + target` 실행으로 처리한다. 테스트 파라미터는 `configs/edge.yaml`, `configs/node.yaml`에 하드코딩한다. -**전제 조건**: node 호스트의 PATH에 `claude` 명령이 있어야 한다. +**전제 조건**: `configs/edge.yaml`의 `console.target`이 가리키는 CLI profile command가 node 호스트에서 실행 가능해야 한다. 기본 예시는 `opencode`이며, `claude`, `gemini`, `codex`, `cline-dgx` 등으로 바꿀 수 있다. -같은 `cli` 어댑터 안에 `gemini`, `codex`, `opencode` profile도 포함할 수 있으며, 각각 `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`, `opencode run --model ollama-dgx/qwen3.6:35b-a3b-bf16 --format default --dangerously-skip-permissions `처럼 headless 조합으로 설정한다. Claude는 `claude -p --dangerously-skip-permissions --output-format stream-json --include-partial-messages --verbose ` 형태로 호출되어 token-level streaming이 활성화된다. +같은 `cli` 어댑터 안에 `claude`, `gemini`, `codex`, `opencode`, `cline` profile을 포함할 수 있다. 예를 들어 `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`, `cline -y --json --config /config/.cline/profiles/ollama-dgx `처럼 headless 조합으로 설정한다. `opencode`는 `mode: "opencode-sse"` profile로 `opencode serve`의 HTTP/SSE 인터페이스를 사용하며, args의 `--model`, `--dangerously-skip-permissions`, `--title` 등은 그대로 유지하면 된다. 실행 순서: -1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. `console.adapter=cli`, `console.agent=claude`, `console.session_id=default`가 기본값이다. +1. edge 호스트에서 `configs/edge.yaml`의 `server.listen`, `nodes[].token`을 확인한다. 예시 설정은 `console.adapter=cli`, `console.target=opencode`, `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에서 받은 입력을 `claude -p --dangerously-skip-permissions --output-format stream-json ... ` one-shot process로 실행한다. +5. node 호스트에서 `./bin/node.sh` — edge 포트가 열릴 때까지 최대 30초 기다린 뒤, edge에서 받은 입력을 선택된 `adapter + target`으로 실행한다. 6. edge 콘솔에서 `/nodes`로 node 등록을 확인한다. 7. (선택 사항) node가 여러 대일 경우 `/node `로 대상을 선택한다. 1대일 경우 자동 선택된다. 8. edge 콘솔의 `edge>` 프롬프트에 메시지를 입력한다. -9. `[node-{alias}-event]` 라인으로 실행 상태를, `[node-{alias}-message]` 라인으로 Codex 응답을 확인한다. +9. `[node-{alias}-event]` 라인으로 실행 상태를, `[node-{alias}-message]` 라인으로 adapter output을 확인한다. 10. edge 콘솔에서 `/exit` 또는 `quit`로 종료한다. 예상 출력: ```text edge> hello -[edge] sent run_id=manual-... node=local-node adapter=cli agent=claude session=default background=false +[edge] sent run_id=manual-... node=local-node adapter=cli target=opencode 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] +[node-local-node-event] complete run_id=manual-... detail="opencode sse execution complete" +[node-local-node-message] ``` ## Console 명령 @@ -47,8 +48,8 @@ edge> hello | `/node ` | 요청을 보낼 명시적 node 선택 | | `/session ` | 현재 console이 사용할 logical session 변경 | | `/background on\|off` | background 실행 모드 토글 (on: 응답 기다리지 않음) | -| `/terminate-session` | 현재 `adapter/agent/session_id`의 worker process 종료 | -| `/status` | 현재 선택된 agent/profile의 사용량(Usage Status) 조회 | +| `/terminate-session` | 현재 `adapter/target/session_id`의 worker process 종료 | +| `/status` | 현재 선택된 target/profile의 사용량(Usage Status) 조회 | | `/exit` | 콘솔 종료 | ## 멀티포인트 라우팅 (Multi-Point Routing) @@ -68,3 +69,5 @@ edge-node transport 연결은 **node id당 1개** TCP 연결만 유지한다. - **terminate session**: session process 명시 종료 (`/terminate-session` 또는 `CancelAction_TERMINATE_SESSION`) 프롬프트 없이 TCP/protobuf edge 서버만 실행하려면 기존처럼 `go run ./apps/edge/cmd/edge serve -c configs/edge.yaml`를 사용한다. + +`bin/node.sh`의 edge 대기 시간은 `IOP_NODE_WAIT_TIMEOUT`, 확인 간격은 `IOP_NODE_WAIT_INTERVAL`, 접속 주소는 `IOP_EDGE_ADDR` 환경 변수로 임시 override할 수 있다. diff --git a/apps/edge/cmd/edge/console.go b/apps/edge/cmd/edge/console.go index 192b0f5..3667fae 100644 --- a/apps/edge/cmd/edge/console.go +++ b/apps/edge/cmd/edge/console.go @@ -50,18 +50,18 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i } }() - // Console state: target node, adapter, agent, session, background mode. + // Console state: target node, adapter execution target, session, background mode. target := &consoleTarget{ Adapter: cfg.Console.Adapter, - Agent: cfg.Console.ResolveAgent(), + Target: cfg.Console.ResolveTarget(), SessionID: normalizeConsoleSessionID(cfg.Console.SessionID), Background: cfg.Console.Background, TimeoutSec: cfg.Console.TimeoutSec, } fmt.Fprintf(out, "IOP Edge console listening on %s\n", cfg.Server.Listen) - fmt.Fprintf(out, "Console target node=%s adapter=%s agent=%s session=%s background=%v\n", - target.NodeRef, target.Adapter, target.Agent, target.SessionID, target.Background) + fmt.Fprintf(out, "Console target node=%s adapter=%s target=%s session=%s background=%v\n", + target.NodeRef, target.Adapter, target.Target, target.SessionID, target.Background) fmt.Fprintln(out, "Start node.sh on another host, then type a message here.") fmt.Fprintln(out, "Commands: /nodes, /node , /session , /background on|off, /terminate-session, /status, /exit") @@ -136,7 +136,7 @@ func runConsole(ctx context.Context, cfg *config.EdgeConfig, in io.Reader, out i type consoleTarget struct { NodeRef string Adapter string - Agent string + Target string SessionID string Background bool TimeoutSec int @@ -163,8 +163,8 @@ func printNodes(out io.Writer, registry *edgenode.Registry, selectedRef string) // buildRunRequest constructs the RunRequest for a console send. // RunRequest.target carries the adapter execution target — for the CLI adapter -// this is the selected agent/profile name. -func buildRunRequest(adapter, agent, sessionID string, background bool, timeoutSec int, message string) (*iop.RunRequest, string, error) { +// this is the selected profile name. +func buildRunRequest(adapter, targetName, sessionID string, background bool, timeoutSec int, message string) (*iop.RunRequest, string, error) { input, err := structpb.NewStruct(map[string]any{"prompt": message}) if err != nil { return nil, "", err @@ -176,7 +176,7 @@ func buildRunRequest(adapter, agent, sessionID string, background bool, timeoutS req := &iop.RunRequest{ RunId: runID, Adapter: adapter, - Target: agent, + Target: targetName, SessionId: normalizeConsoleSessionID(sessionID), SessionMode: iop.RunSessionMode_RUN_SESSION_MODE_CREATE_IF_MISSING, Background: background, @@ -195,7 +195,7 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *co return err } - req, runID, err := buildRunRequest(target.Adapter, target.Agent, target.SessionID, target.Background, target.TimeoutSec, message) + req, runID, err := buildRunRequest(target.Adapter, target.Target, target.SessionID, target.Background, target.TimeoutSec, message) if err != nil { return err } @@ -211,8 +211,8 @@ func sendConsoleRun(ctx context.Context, registry *edgenode.Registry, events *co if nodeAlias == "" { nodeAlias = entry.NodeID } - fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s agent=%s session=%s background=%v\n", - runID, nodeAlias, target.Adapter, target.Agent, req.GetSessionId(), target.Background) + fmt.Fprintf(out, "[edge] sent run_id=%s node=%s adapter=%s target=%s session=%s background=%v\n", + runID, nodeAlias, target.Adapter, target.Target, req.GetSessionId(), target.Background) if err := entry.Client.Send(req); err != nil { return err } @@ -283,7 +283,7 @@ func sendTerminateSession(ctx context.Context, registry *edgenode.Registry, targ } req := &iop.CancelRequest{ Adapter: target.Adapter, - Target: target.Agent, + Target: target.Target, SessionId: normalizeConsoleSessionID(target.SessionID), Action: iop.CancelAction_CANCEL_ACTION_TERMINATE_SESSION, } @@ -301,7 +301,7 @@ func normalizeConsoleSessionID(id string) string { return id } -func buildNodeCommandRequest(adapter, agent, sessionID string, timeoutSec int) (*iop.NodeCommandRequest, string) { +func buildNodeCommandRequest(adapter, targetName, sessionID string, timeoutSec int) (*iop.NodeCommandRequest, string) { if timeoutSec <= 0 { timeoutSec = 30 } @@ -310,7 +310,7 @@ func buildNodeCommandRequest(adapter, agent, sessionID string, timeoutSec int) ( RequestId: reqID, Type: iop.NodeCommandType_NODE_COMMAND_TYPE_USAGE_STATUS, Adapter: adapter, - Target: agent, + Target: targetName, SessionId: normalizeConsoleSessionID(sessionID), TimeoutSec: int32(timeoutSec), } @@ -327,13 +327,13 @@ func sendConsoleStatus(ctx context.Context, registry *edgenode.Registry, out io. return err } - req, _ := buildNodeCommandRequest(target.Adapter, target.Agent, target.SessionID, target.TimeoutSec) + req, _ := buildNodeCommandRequest(target.Adapter, target.Target, target.SessionID, target.TimeoutSec) nodeAlias := entry.Alias if nodeAlias == "" { nodeAlias = entry.NodeID } - fmt.Fprintf(out, "[edge] sent command=status node=%s adapter=%s agent=%s session=%s\n", nodeAlias, target.Adapter, target.Agent, req.GetSessionId()) + fmt.Fprintf(out, "[edge] sent command=status node=%s adapter=%s target=%s session=%s\n", nodeAlias, target.Adapter, target.Target, req.GetSessionId()) timeout := statusWaitTimeout(req) resp, err := toki.SendRequestTyped[*iop.NodeCommandRequest, *iop.NodeCommandResponse]( @@ -349,12 +349,12 @@ func sendConsoleStatus(ctx context.Context, registry *edgenode.Registry, out io. return fmt.Errorf("node reported error: %s", resp.GetError()) } - formatUsageStatus(out, nodeAlias, target.Agent, req.GetSessionId(), resp.GetUsageStatus()) + formatUsageStatus(out, nodeAlias, target.Target, req.GetSessionId(), resp.GetUsageStatus()) return nil } -func formatUsageStatus(out io.Writer, nodeAlias, agent, sessionID string, status *iop.AgentUsageStatus) { - fmt.Fprintf(out, "[node-%s-status] agent=%s session=%s\n", nodeAlias, agent, sessionID) +func formatUsageStatus(out io.Writer, nodeAlias, targetName, sessionID string, status *iop.AgentUsageStatus) { + fmt.Fprintf(out, "[node-%s-status] target=%s session=%s\n", nodeAlias, targetName, sessionID) if status == nil { fmt.Fprintln(out, "no usage status provided") diff --git a/apps/edge/cmd/edge/main.go b/apps/edge/cmd/edge/main.go index 0f0b7b6..49ec3e3 100644 --- a/apps/edge/cmd/edge/main.go +++ b/apps/edge/cmd/edge/main.go @@ -24,7 +24,7 @@ func main() { func rootCmd() *cobra.Command { root := &cobra.Command{ Use: "edge", - Short: "IOP Edge — node gateway server", + Short: "IOP Edge — execution group controller", } root.PersistentFlags().StringVarP(&cfgFile, "config", "c", "configs/edge.yaml", "config file path") root.AddCommand(serveCmd(), consoleCmd()) diff --git a/apps/edge/internal/node/mapper_test.go b/apps/edge/internal/node/mapper_test.go index 7e67aa0..7915928 100644 --- a/apps/edge/internal/node/mapper_test.go +++ b/apps/edge/internal/node/mapper_test.go @@ -137,6 +137,57 @@ func TestBuildConfigPayload_CLIProfiles(t *testing.T) { } } +func TestBuildConfigPayload_CLIProfileOpencodeSSEMode(t *testing.T) { + rec := &edgenode.NodeRecord{ + ID: "node-1", + Alias: "test", + Token: "token", + Adapters: config.AdaptersConf{ + CLI: config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": { + Command: "/usr/local/bin/opencode", + Args: []string{"--model", "m"}, + Mode: "opencode-sse", + }, + }, + }, + }, + Runtime: config.RuntimeConf{WorkspaceRoot: "/tmp/ws"}, + } + + payload, err := edgenode.BuildConfigPayload(rec) + if err != nil { + t.Fatalf("BuildConfigPayload failed: %v", err) + } + + var found bool + for _, a := range payload.Adapters { + if a.Type != "cli" { + continue + } + cli := a.GetCli() + if cli == nil { + t.Fatal("expected cli config") + } + p, ok := cli.Profiles["opencode"] + if !ok { + t.Fatal("expected opencode profile in payload") + } + if p.Mode != "opencode-sse" { + t.Errorf("mode: got %q, want %q", p.Mode, "opencode-sse") + } + if len(p.Args) != 2 || p.Args[0] != "--model" || p.Args[1] != "m" { + t.Errorf("args: got %v", p.Args) + } + found = true + } + if !found { + t.Fatal("expected cli adapter in payload") + } +} + func TestBuildConfigPayload_MockAlwaysPresent(t *testing.T) { rec := &edgenode.NodeRecord{ ID: "node-1", diff --git a/apps/node/README.md b/apps/node/README.md index 23df2e6..6a81162 100644 --- a/apps/node/README.md +++ b/apps/node/README.md @@ -1,7 +1,7 @@ # node — Node Agent 디바이스당 1개 실행되는 IOP 노드 에이전트. -실제 AI 모델 추론 어댑터를 관리하고, TCP + Protobuf + mTLS 소켓을 통해 요청을 처리한다. +Edge에 연결되어 `adapter + target` 실행을 수행하고, TCP/protobuf 소켓을 통해 요청, 명령, 이벤트 스트림을 처리한다. mTLS helper는 `packages/auth`에 있지만 현재 Edge-Node transport에는 아직 연결하지 않았다. ## 내부 구조 (Hexagonal Architecture) @@ -12,7 +12,7 @@ internal/ node/ — 핵심 노드 서비스 (transport.Handler 구현) runtime/ — 도메인 타입: ExecutionSpec, RuntimeEvent, 인터페이스 router/ — RunRequest → ExecutionSpec 라우팅 - transport/ — TCP 서버, 세션 관리, 프레임 코덱 + transport/ — Edge로 연결하는 TCP/protobuf client와 session 처리 adapters/ mock/ — 에코 테스트 어댑터 ollama/ — Ollama API 어댑터 (TODO) @@ -43,15 +43,17 @@ go build -o bin/node ./apps/node/cmd/node ./bin/node.sh ``` -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). +`bin/node.sh`는 `configs/node.yaml`의 `transport.edge_addr` 포트가 열릴 때까지 최대 30초 기다린 뒤 node를 시작한다. 테스트 중 edge 주소만 임시로 바꾸려면 `IOP_EDGE_ADDR=host:port ./bin/node.sh`를 사용할 수 있다. -edge에서 실행 요청을 받으면 node는 해당 입력을 Claude process의 prompt 인자로 전달하고, stream-json delta를 토큰 단위로 emit한다. +edge 예시 설정은 cli adapter의 `opencode` profile을 사용한다. `configs/edge.yaml`의 `console.target`을 `claude`, `gemini`, `codex`, `opencode`, `cline-dgx` 같은 profile 이름으로 바꾸면 같은 실행 파이프라인에서 다른 CLI target을 검증할 수 있다. + +edge에서 실행 요청을 받으면 node는 해당 입력을 선택된 adapter target으로 전달하고, adapter가 emit한 delta/event를 Edge로 스트리밍한다. ```text [edge-message] hello [node-event] start run_id=manual-... -[node-event] complete run_id=manual-... detail="cli execution complete" -[node-message] +[node-event] complete run_id=manual-... detail="opencode sse execution complete" +[node-message] ``` ## Logical Session (transport 1개 · session 여러 개) @@ -61,7 +63,7 @@ edge-node transport 연결은 **호스트당 1개**를 유지한다. 그 연결 | 개념 | 설명 | |---|---| | transport 연결 | edge-node 호스트 쌍당 1개 TCP 연결 | -| logical session | `(adapter, agent, session_id)` 로 식별되는 장수 worker process | +| logical session | `(adapter, target, session_id)` 로 식별되는 장수 worker process | | run | session 위에서 실행되는 단일 요청 | **cancel vs terminate:** @@ -76,9 +78,9 @@ edge-node transport 연결은 **호스트당 1개**를 유지한다. 그 연결 ## Node Commands -node는 모델 실행(`RunRequest`) 외에도 edge가 보내는 `NodeCommandRequest`를 처리한다. +node는 adapter execution(`RunRequest`) 외에도 edge가 보내는 `NodeCommandRequest`를 처리한다. 현재 구현된 command: -- `USAGE_STATUS`: 선택된 `adapter/agent` (예: `cli/codex`)의 사용량 한도와 초기화 시간을 조회한다. +- `USAGE_STATUS`: 선택된 `adapter/target` (예: `cli/codex`)의 사용량 한도와 초기화 시간을 조회한다. - Codex의 경우 TUI를 시작하고 `/status` 명령을 전송한 뒤 출력된 `% left` 정보를 파싱해 `AgentUsageStatus`로 반환한다. ## 어댑터 @@ -90,13 +92,15 @@ node는 모델 실행(`RunRequest`) 외에도 edge가 보내는 `NodeCommandRequ | `vllm` | vLLM OpenAI-compatible API | TODO | | `cli` | claude/gemini/codex/opencode/cline CLI 실행 | 구현 완료 | -`claude`, `gemini`, `codex`, `opencode`, `cline`처럼 기본이 interactive TUI인 CLI는 이 파이프라인에서 non-interactive 모드로 설정해야 한다. 예: `claude -p --dangerously-skip-permissions`, `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`, `opencode run --model ollama-dgx/qwen3.6:35b-a3b-bf16 --format json --dangerously-skip-permissions `, `cline -y --json --config /config/.cline/profiles/ollama-dgx `. +`claude`, `gemini`, `codex`, `opencode`, `cline`처럼 기본이 interactive TUI인 CLI는 이 파이프라인에서 non-interactive 모드로 설정해야 한다. 예: `claude -p --dangerously-skip-permissions`, `gemini --approval-mode yolo -p `, `codex exec --dangerously-bypass-approvals-and-sandbox`, `cline -y --json --config /config/.cline/profiles/ollama-dgx `. + +`opencode`는 profile `mode: "opencode-sse"`로 설정해 `opencode serve`의 HTTP/SSE 인터페이스를 사용한다. `profile.Command`가 가리키는 opencode 바이너리를 `serve --hostname 127.0.0.1 --port 0`으로 띄우고 `/global/event` SSE에서 text delta를 `RuntimeEvent` delta로 relay한다. 외부에서 이미 실행 중인 server에 연결하려면 `--attach `을 args에 추가한다. legacy `output_format: "opencode-json"` stdout JSONL 경로는 `opencode run --format json` 사용 시에만 의미가 있다. Cline은 mutable `--config` 디렉터리를 프로필처럼 나눈다. `/config/.cline/profiles/ollama-dgx`와 `/config/.cline/profiles/ollama-m1`는 `opencode`의 `ollama-dgx`/`ollama-m1` provider 설정을 Cline의 `ollama` provider 설정으로 변환한 값이다. ## Transport - TCP 4-byte length-prefix framing -- JSON-encoded Envelope (TODO: protobuf 교체) -- mTLS (configs/node.yaml에서 활성화) +- protobuf message framing via `common-proto-socket` +- mTLS helper는 존재하지만 현재 transport 설정에는 아직 연결되지 않음 - 하트비트: 30초 간격 diff --git a/apps/node/cmd/node/main.go b/apps/node/cmd/node/main.go index ba3f335..fa99a8d 100644 --- a/apps/node/cmd/node/main.go +++ b/apps/node/cmd/node/main.go @@ -25,7 +25,7 @@ func main() { func rootCmd() *cobra.Command { root := &cobra.Command{ Use: "node", - Short: "IOP Node Agent — runs inference adapters on this device", + Short: "IOP Node Agent — runs adapter executions on this device", } root.PersistentFlags().StringVarP(&cfgFile, "config", "c", "configs/node.yaml", "config file path") diff --git a/apps/node/internal/adapters/cli/cli.go b/apps/node/internal/adapters/cli/cli.go index acca9b2..35debbb 100644 --- a/apps/node/internal/adapters/cli/cli.go +++ b/apps/node/internal/adapters/cli/cli.go @@ -1,8 +1,9 @@ // Package cli provides an Adapter that runs external CLI tools as an execution -// adapter. Profiles (claude, gemini, codex, opencode, cline) are configured in -// configs/node.yaml and select the command + args to execute. Interactive CLIs -// should be configured with their non-interactive/headless flags for use in the -// request/response node pipeline. +// adapter. Profiles (claude, gemini, codex, opencode, cline) are defined by +// edge configuration, pushed to the node in NodeConfigPayload, and selected as +// the adapter execution target. Interactive CLIs should be configured with +// their non-interactive/headless flags for use in the request/response node +// pipeline. package cli import ( @@ -23,6 +24,11 @@ import ( const Name = "cli" +const ( + modeCodexExec = "codex-exec" + modeOpencodeSSE = "opencode-sse" +) + type cliOutput struct { line string } @@ -52,11 +58,12 @@ type codexExecSession struct { } type CLI struct { - mu sync.Mutex - profiles map[string]config.CLIProfileConf - sessions map[sessionKey]*profileSession - codexSessions map[sessionKey]*codexExecSession - logger *zap.Logger + mu sync.Mutex + profiles map[string]config.CLIProfileConf + sessions map[sessionKey]*profileSession + codexSessions map[sessionKey]*codexExecSession + opencodeSessions map[sessionKey]*opencodeSSESession + logger *zap.Logger // StatusChecker overrides status.CheckUsage for testing. StatusChecker func(ctx context.Context, agent string, profile config.CLIProfileConf) (*status.UsageStatus, error) @@ -64,10 +71,11 @@ type CLI struct { func New(cfg config.CLIConf, logger *zap.Logger) *CLI { return &CLI{ - profiles: cfg.Profiles, - sessions: make(map[sessionKey]*profileSession), - codexSessions: make(map[sessionKey]*codexExecSession), - logger: logger, + profiles: cfg.Profiles, + sessions: make(map[sessionKey]*profileSession), + codexSessions: make(map[sessionKey]*codexExecSession), + opencodeSessions: make(map[sessionKey]*opencodeSSESession), + logger: logger, } } @@ -91,7 +99,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 && c.profiles[name].Mode != "codex-exec" { + if c.profiles[name].Persistent && c.profiles[name].Mode != modeCodexExec && c.profiles[name].Mode != modeOpencodeSSE { names = append(names, name) } } @@ -138,8 +146,17 @@ func (c *CLI) Stop(_ context.Context) error { } c.sessions = make(map[sessionKey]*profileSession) c.codexSessions = make(map[sessionKey]*codexExecSession) + opencodeCopy := make(map[sessionKey]*opencodeSSESession, len(c.opencodeSessions)) + for key, sess := range c.opencodeSessions { + opencodeCopy[key] = sess + } + c.opencodeSessions = make(map[sessionKey]*opencodeSSESession) c.mu.Unlock() + for _, sess := range opencodeCopy { + closeOpencodeSession(sess) + } + var firstErr error for key, sess := range sessionsCopy { if err := closeProfileSession(context.Background(), sess); err != nil && firstErr == nil { @@ -155,9 +172,12 @@ func (c *CLI) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runt if !ok { return fmt.Errorf("cli adapter: unknown agent %q", agent) } - if profile.Mode == "codex-exec" { + if profile.Mode == modeCodexExec { return c.executeCodexExec(ctx, spec, profile, sink) } + if profile.Mode == modeOpencodeSSE { + return c.executeOpencodeSSE(ctx, spec, profile, sink) + } if profile.Persistent { return c.executePersistent(ctx, spec, profile, sink) } @@ -194,7 +214,7 @@ func (c *CLI) HandleCommand(ctx context.Context, req runtime.CommandRequest) (ru // TerminateSession implements runtime.SessionTerminator. func (c *CLI) TerminateSession(_ context.Context, agent, sessionID string) error { key := sessionKey{agent: agent, sessionID: normalizeSessionID(sessionID)} - if profile, ok := c.profiles[agent]; ok && profile.Mode == "codex-exec" { + if profile, ok := c.profiles[agent]; ok && profile.Mode == modeCodexExec { c.mu.Lock() _, ok := c.codexSessions[key] if ok { @@ -206,6 +226,19 @@ func (c *CLI) TerminateSession(_ context.Context, agent, sessionID string) error } return nil } + if profile, ok := c.profiles[agent]; ok && profile.Mode == modeOpencodeSSE { + c.mu.Lock() + sess, ok := c.opencodeSessions[key] + if ok { + delete(c.opencodeSessions, key) + } + c.mu.Unlock() + if !ok { + return fmt.Errorf("cli adapter: no session %q for agent %q", key.sessionID, agent) + } + closeOpencodeSession(sess) + return nil + } c.mu.Lock() sess, ok := c.sessions[key] diff --git a/apps/node/internal/adapters/cli/opencode_sse.go b/apps/node/internal/adapters/cli/opencode_sse.go new file mode 100644 index 0000000..4a44d1a --- /dev/null +++ b/apps/node/internal/adapters/cli/opencode_sse.go @@ -0,0 +1,737 @@ +package cli + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os/exec" + "regexp" + "strings" + "sync" + "time" + + "go.uber.org/zap" + + "iop/apps/node/internal/runtime" + "iop/packages/config" +) + +// opencodeSSESession tracks a logical session backed by an OpenCode server. +// When owned is true, the local server process is started by the adapter and +// must be killed on close. When false, the adapter attaches to an external +// server via --attach and never owns the process lifecycle. +type opencodeSSESession struct { + key sessionKey + serverURL string + sessionID string + cmd *exec.Cmd + owned bool + mu sync.Mutex +} + +// opencodeRunOpts is the subset of opencode CLI options that affect SSE mode. +type opencodeRunOpts struct { + AttachURL string + Model string + Agent string + Variant string + Title string + Dir string + DangerouslySkipPermissions bool +} + +func parseOpencodeRunArgs(args []string) opencodeRunOpts { + var opts opencodeRunOpts + for i := 0; i < len(args); i++ { + a := args[i] + next := func() (string, bool) { + if i+1 < len(args) { + v := args[i+1] + i++ + return v, true + } + return "", false + } + switch a { + case "--attach": + if v, ok := next(); ok { + opts.AttachURL = v + } + case "--model": + if v, ok := next(); ok { + opts.Model = v + } + case "--agent": + if v, ok := next(); ok { + opts.Agent = v + } + case "--variant": + if v, ok := next(); ok { + opts.Variant = v + } + case "--title": + if v, ok := next(); ok { + opts.Title = v + } + case "--dir": + if v, ok := next(); ok { + opts.Dir = v + } + case "--dangerously-skip-permissions": + opts.DangerouslySkipPermissions = true + } + } + return opts +} + +// decodeSSEDataLine returns the JSON payload of an SSE "data:" line, or +// (nil, false) if the line is not a usable data line. +func decodeSSEDataLine(line []byte) ([]byte, bool) { + if !bytes.HasPrefix(line, []byte("data:")) { + return nil, false + } + data := bytes.TrimSpace(line[len("data:"):]) + if len(data) == 0 { + return nil, false + } + return data, true +} + +func (c *CLI) executeOpencodeSSE(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { + opts := parseOpencodeRunArgs(profile.Args) + + sess, err := c.resolveOpencodeSession(ctx, spec, profile, opts) + if err != nil { + return err + } + + sess.mu.Lock() + defer sess.mu.Unlock() + + // Open SSE stream before sending the prompt so early events are not missed. + // OpenCode 1.14.x publishes session events through /global/event as a + // top-level payload wrapper; /event only reports server connection events. + sseReq, err := http.NewRequestWithContext(ctx, http.MethodGet, sess.serverURL+"/global/event", nil) + if err != nil { + return fmt.Errorf("cli adapter: opencode sse new request: %w", err) + } + sseReq.Header.Set("Accept", "text/event-stream") + sseResp, err := http.DefaultClient.Do(sseReq) + if err != nil { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: fmt.Sprintf("opencode sse connect: %v", err), + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: opencode sse connect: %w", err) + } + defer sseResp.Body.Close() + if sseResp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(sseResp.Body, 1024)) + msg := fmt.Sprintf("opencode sse status %d: %s", sseResp.StatusCode, strings.TrimSpace(string(body))) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: msg, + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: %s", msg) + } + + if sess.sessionID == "" { + id, err := opencodeCreateSession(ctx, sess.serverURL, opts.Title) + if err != nil { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: fmt.Sprintf("opencode create session: %v", err), + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: opencode create session: %w", err) + } + sess.sessionID = id + } + + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeStart, + Timestamp: time.Now(), + }) + + prompt := extractPrompt(spec.Input) + if err := opencodePromptAsync(ctx, sess.serverURL, sess.sessionID, prompt, opts); err != nil { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: fmt.Sprintf("opencode prompt async: %v", err), + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: opencode prompt async: %w", err) + } + + return c.driveOpencodeSSE(ctx, spec, sess, opts, sseResp.Body, sink) +} + +func (c *CLI) resolveOpencodeSession(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, opts opencodeRunOpts) (*opencodeSSESession, error) { + agent := cliAgentName(spec) + key := sessionKey{agent: agent, sessionID: normalizeSessionID(spec.SessionID)} + + c.mu.Lock() + if sess, ok := c.opencodeSessions[key]; ok { + c.mu.Unlock() + return sess, nil + } + if spec.SessionMode == runtime.SessionModeRequireExisting { + c.mu.Unlock() + return nil, fmt.Errorf("cli adapter: no persistent session for agent %q session %q", agent, key.sessionID) + } + c.mu.Unlock() + + serverURL := strings.TrimRight(opts.AttachURL, "/") + var cmd *exec.Cmd + owned := false + if serverURL == "" { + url, started, err := startOpencodeLocalServer(ctx, profile, c.logger) + if err != nil { + return nil, fmt.Errorf("cli adapter: start opencode server: %w", err) + } + serverURL = url + cmd = started + owned = true + } + + sess := &opencodeSSESession{ + key: key, + serverURL: serverURL, + cmd: cmd, + owned: owned, + } + + c.mu.Lock() + if existing, ok := c.opencodeSessions[key]; ok { + c.mu.Unlock() + if owned && cmd != nil && cmd.Process != nil { + _ = cmd.Process.Kill() + } + return existing, nil + } + c.opencodeSessions[key] = sess + c.mu.Unlock() + return sess, nil +} + +var opencodeServerListenRE = regexp.MustCompile(`(?i)opencode server listening on (https?://\S+)`) + +func startOpencodeLocalServer(ctx context.Context, profile config.CLIProfileConf, logger *zap.Logger) (string, *exec.Cmd, error) { + args := []string{"serve", "--hostname", "127.0.0.1", "--port", "0"} + cmd := exec.Command(profile.Command, args...) + if len(profile.Env) > 0 { + cmd.Env = append(cmd.Environ(), profile.Env...) + } + + stdout, err := cmd.StdoutPipe() + if err != nil { + return "", nil, fmt.Errorf("stdout pipe: %w", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return "", nil, fmt.Errorf("stderr pipe: %w", err) + } + if err := cmd.Start(); err != nil { + return "", nil, fmt.Errorf("start opencode serve: %w", err) + } + + urlCh := make(chan string, 2) + scan := func(r io.Reader, name string) { + s := bufio.NewScanner(r) + s.Buffer(make([]byte, 64*1024), 1024*1024) + for s.Scan() { + line := s.Text() + if logger != nil { + logger.Debug("opencode server log", zap.String("stream", name), zap.String("line", line)) + } + if m := opencodeServerListenRE.FindStringSubmatch(line); len(m) >= 2 { + select { + case urlCh <- m[1]: + default: + } + } + } + } + go scan(stdout, "stdout") + go scan(stderr, "stderr") + + select { + case u := <-urlCh: + return strings.TrimRight(u, "/"), cmd, nil + case <-time.After(30 * time.Second): + _ = cmd.Process.Kill() + return "", nil, errors.New("timeout waiting for opencode server listen line") + case <-ctx.Done(): + _ = cmd.Process.Kill() + return "", nil, ctx.Err() + } +} + +type opencodeEnvelope struct { + Type string `json:"type"` + Properties map[string]any `json:"properties"` +} + +type opencodeGlobalEnvelope struct { + Payload *opencodeEnvelope `json:"payload"` +} + +func opencodeEventEnvelope(data []byte) (opencodeEnvelope, bool) { + var env opencodeEnvelope + if err := json.Unmarshal(data, &env); err == nil && env.Type != "" && env.Type != "sync" { + return env, true + } + var global opencodeGlobalEnvelope + if err := json.Unmarshal(data, &global); err != nil || global.Payload == nil || global.Payload.Type == "" || global.Payload.Type == "sync" { + return opencodeEnvelope{}, false + } + return *global.Payload, true +} + +func (c *CLI) driveOpencodeSSE(ctx context.Context, spec runtime.ExecutionSpec, sess *opencodeSSESession, opts opencodeRunOpts, body io.Reader, sink runtime.EventSink) error { + type evt struct { + ev opencodeEnvelope + raw []byte + } + events := make(chan evt, 32) + scanErrCh := make(chan error, 1) + + go func() { + defer close(events) + s := bufio.NewScanner(body) + s.Buffer(make([]byte, 64*1024), 4*1024*1024) + for s.Scan() { + line := s.Bytes() + data, ok := decodeSSEDataLine(line) + if !ok { + continue + } + env, ok := opencodeEventEnvelope(data) + if !ok { + continue + } + cp := make([]byte, len(data)) + copy(cp, data) + events <- evt{ev: env, raw: cp} + } + scanErrCh <- s.Err() + }() + + var ( + inputTokens int + outputTokens int + partTypes = map[string]string{} + ) + + finalize := func(reason error) error { + bg, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if sess.sessionID != "" { + _ = opencodeAbort(bg, sess.serverURL, sess.sessionID) + } + _ = sink.Emit(context.Background(), runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeCancelled, + Message: cancelEventForContext(reason), + Timestamp: time.Now(), + }) + return runtime.ErrRunCancelled + } + + for { + select { + case <-ctx.Done(): + return finalize(ctx.Err()) + case e, ok := <-events: + if !ok { + if ctx.Err() != nil { + return finalize(ctx.Err()) + } + if err := <-scanErrCh; err != nil && !errors.Is(err, io.EOF) { + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: fmt.Sprintf("opencode sse scan: %v", err), + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: opencode sse scan: %w", err) + } + return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) + } + + // Filter events that carry a sessionID mismatching ours, when present. + if sid, ok := stringFromProp(e.ev.Properties, "sessionID"); ok && sid != "" && sess.sessionID != "" && sid != sess.sessionID { + continue + } + + switch e.ev.Type { + case "session.next.text.delta": + delta := opencodeDeltaText(e.ev.Properties) + if delta != "" { + outputTokens += len(strings.Fields(delta)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeDelta, + Delta: delta, + Timestamp: time.Now(), + }) + } + case "message.part.updated": + if partID, partType, ok := opencodePartInfo(e.ev.Properties); ok { + partTypes[partID] = partType + } + if in, out, ok := opencodeStepTokens(e.ev.Properties); ok { + if in > 0 { + inputTokens = in + } + if out > 0 { + outputTokens = out + } + } + case "message.part.delta": + delta := opencodeMessagePartDeltaText(e.ev.Properties, partTypes) + if delta != "" { + outputTokens += len(strings.Fields(delta)) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeDelta, + Delta: delta, + Timestamp: time.Now(), + }) + } + case "session.next.step.ended": + if in, out, ok := opencodeStepTokens(e.ev.Properties); ok { + if in > 0 { + inputTokens = in + } + if out > 0 { + outputTokens = out + } + } + case "session.idle": + return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) + case "session.status": + if opencodeStatusIdle(e.ev.Properties) { + return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) + } + case "session.error": + msg := opencodeErrorMessage(e.ev.Properties) + _ = sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeError, + Error: msg, + Timestamp: time.Now(), + }) + return fmt.Errorf("cli adapter: opencode session error: %s", msg) + case "permission.asked": + permID, _ := stringFromProp(e.ev.Properties, "id") + if permID == "" { + permID, _ = stringFromProp(e.ev.Properties, "permissionID") + } + if permID != "" { + reply := "reject" + if opts.DangerouslySkipPermissions { + reply = "once" + } + bg, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = opencodePermissionReply(bg, sess.serverURL, permID, reply) + cancel() + } + } + } + } +} + +func stringFromProp(props map[string]any, key string) (string, bool) { + if props == nil { + return "", false + } + v, ok := props[key] + if !ok { + return "", false + } + s, ok := v.(string) + return s, ok +} + +func opencodeDeltaText(props map[string]any) string { + if props == nil { + return "" + } + if s, ok := props["delta"].(string); ok && s != "" { + return s + } + if part, ok := props["part"].(map[string]any); ok { + if s, ok := part["text"].(string); ok { + return s + } + } + return "" +} + +func opencodePartInfo(props map[string]any) (string, string, bool) { + if props == nil { + return "", "", false + } + part, ok := props["part"].(map[string]any) + if !ok { + return "", "", false + } + id, _ := part["id"].(string) + partType, _ := part["type"].(string) + if id == "" || partType == "" { + return "", "", false + } + return id, partType, true +} + +func opencodeMessagePartDeltaText(props map[string]any, partTypes map[string]string) string { + if props == nil { + return "" + } + field, _ := props["field"].(string) + if field != "text" { + return "" + } + partID, _ := props["partID"].(string) + if partID == "" || partTypes[partID] != "text" { + return "" + } + delta, _ := props["delta"].(string) + return delta +} + +func opencodeStepTokens(props map[string]any) (int, int, bool) { + if props == nil { + return 0, 0, false + } + tokens, ok := props["tokens"].(map[string]any) + if !ok { + if part, partOK := props["part"].(map[string]any); partOK { + tokens, ok = part["tokens"].(map[string]any) + } + } + if !ok { + return 0, 0, false + } + in := intFromNumber(tokens["input"]) + out := intFromNumber(tokens["output"]) + return in, out, true +} + +func intFromNumber(v any) int { + switch n := v.(type) { + case float64: + return int(n) + case int: + return n + case json.Number: + if i, err := n.Int64(); err == nil { + return int(i) + } + } + return 0 +} + +func emitOpencodeComplete(ctx context.Context, sink runtime.EventSink, runID string, inputTokens, outputTokens int) error { + return sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeComplete, + Message: "opencode sse execution complete", + Usage: &runtime.UsageStats{ + InputTokens: inputTokens, + OutputTokens: outputTokens, + }, + Timestamp: time.Now(), + }) +} + +// opencodeModelPayload parses "provider/model" into the OpenCode prompt_async +// model object {"providerID": provider, "modelID": model}. Returns an error +// when either side is missing. +func opencodeModelPayload(s string) (map[string]any, error) { + idx := strings.Index(s, "/") + if idx < 0 { + return nil, fmt.Errorf("cli adapter: opencode model %q missing provider/model separator", s) + } + provider := s[:idx] + model := s[idx+1:] + if provider == "" || model == "" { + return nil, fmt.Errorf("cli adapter: opencode model %q has empty provider or model", s) + } + return map[string]any{"providerID": provider, "modelID": model}, nil +} + +// opencodeStatusIdle reports whether a session.status event's status payload +// represents idle. OpenCode emits the status as a SessionStatus object +// ({"type":"idle"}); older builds may use the bare string "idle". +func opencodeStatusIdle(props map[string]any) bool { + if props == nil { + return false + } + switch v := props["status"].(type) { + case string: + return v == "idle" + case map[string]any: + t, _ := v["type"].(string) + return t == "idle" + } + return false +} + +func opencodeErrorMessage(props map[string]any) string { + if props == nil { + return "opencode error" + } + if errObj, ok := props["error"].(map[string]any); ok { + if data, ok := errObj["data"].(map[string]any); ok { + if m, ok := data["message"].(string); ok && m != "" { + return m + } + } + if name, ok := errObj["name"].(string); ok && name != "" { + return name + } + } + if m, ok := props["message"].(string); ok && m != "" { + return m + } + return "opencode error" +} + +// --- HTTP helpers --- + +func opencodeCreateSession(ctx context.Context, baseURL, title string) (string, error) { + payload := map[string]any{} + if title != "" { + payload["title"] = title + } + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(payload); err != nil { + return "", err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session", &body) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return "", fmt.Errorf("opencode create session status %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) + } + var raw map[string]any + if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { + return "", fmt.Errorf("decode session response: %w", err) + } + if id, ok := raw["id"].(string); ok && id != "" { + return id, nil + } + if s, ok := raw["session"].(map[string]any); ok { + if id, ok := s["id"].(string); ok && id != "" { + return id, nil + } + } + return "", errors.New("opencode create session: missing id in response") +} + +func opencodePromptAsync(ctx context.Context, baseURL, sessionID, prompt string, opts opencodeRunOpts) error { + payload := map[string]any{ + "parts": []map[string]any{ + {"type": "text", "text": prompt}, + }, + } + if opts.Model != "" { + model, err := opencodeModelPayload(opts.Model) + if err != nil { + return err + } + payload["model"] = model + } + if opts.Agent != "" { + payload["agent"] = opts.Agent + } + if opts.Variant != "" { + payload["variant"] = opts.Variant + } + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(payload); err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session/"+sessionID+"/prompt_async", &body) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) + } + _, _ = io.Copy(io.Discard, resp.Body) + return nil +} + +func opencodePermissionReply(ctx context.Context, baseURL, permID, reply string) error { + var body bytes.Buffer + if err := json.NewEncoder(&body).Encode(map[string]any{"reply": reply}); err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/permission/"+permID+"/reply", &body) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + return nil +} + +func opencodeAbort(ctx context.Context, baseURL, sessionID string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session/"+sessionID+"/abort", nil) + if err != nil { + return err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + return nil +} + +// closeOpencodeSession kills the owned local server process if any. For +// attached sessions, no process is owned and no kill happens. +func closeOpencodeSession(sess *opencodeSSESession) { + if sess == nil { + return + } + if sess.owned && sess.cmd != nil && sess.cmd.Process != nil { + _ = sess.cmd.Process.Kill() + } +} diff --git a/apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go b/apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go new file mode 100644 index 0000000..96e1e01 --- /dev/null +++ b/apps/node/internal/adapters/cli/opencode_sse_blackbox_test.go @@ -0,0 +1,646 @@ +package cli_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "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" +) + +// opencodeFakeServer is a minimal stand-in for the OpenCode HTTP/SSE server. +// Tests push SSE events through ch; the SSE handler forwards them as +// `data: \n\n` lines until the channel closes. +type opencodeFakeServer struct { + mu sync.Mutex + events chan map[string]any + abortCalls atomic.Int32 + promptCalls atomic.Int32 + createCalls atomic.Int32 + lastPromptBody []byte + createdID string +} + +func newOpencodeFakeServer(t *testing.T, sessionID string) (*opencodeFakeServer, *httptest.Server) { + t.Helper() + s := &opencodeFakeServer{ + events: make(chan map[string]any, 32), + createdID: sessionID, + } + mux := http.NewServeMux() + eventHandler := func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + if flusher != nil { + flusher.Flush() + } + for { + select { + case <-r.Context().Done(): + return + case ev, ok := <-s.events: + if !ok { + return + } + b, err := json.Marshal(ev) + if err != nil { + continue + } + fmt.Fprintf(w, "data: %s\n\n", string(b)) + if flusher != nil { + flusher.Flush() + } + } + } + } + mux.HandleFunc("/event", eventHandler) + mux.HandleFunc("/global/event", eventHandler) + mux.HandleFunc("/session", func(w http.ResponseWriter, r *http.Request) { + s.createCalls.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"id": s.createdID}) + }) + mux.HandleFunc("/session/", func(w http.ResponseWriter, r *http.Request) { + // Routes: /session/{id}/prompt_async or /session/{id}/abort + path := strings.TrimPrefix(r.URL.Path, "/session/") + parts := strings.SplitN(path, "/", 2) + if len(parts) != 2 { + http.NotFound(w, r) + return + } + switch parts[1] { + case "prompt_async": + s.promptCalls.Add(1) + buf := make([]byte, 4096) + n, _ := r.Body.Read(buf) + s.mu.Lock() + s.lastPromptBody = append([]byte(nil), buf[:n]...) + s.mu.Unlock() + w.WriteHeader(http.StatusAccepted) + case "abort": + s.abortCalls.Add(1) + w.WriteHeader(http.StatusOK) + default: + http.NotFound(w, r) + } + }) + mux.HandleFunc("/permission/", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + srv := httptest.NewServer(mux) + t.Cleanup(func() { + s.mu.Lock() + if s.events != nil { + close(s.events) + s.events = nil + } + s.mu.Unlock() + srv.Close() + }) + return s, srv +} + +func (s *opencodeFakeServer) push(ev map[string]any) { + s.mu.Lock() + ch := s.events + s.mu.Unlock() + if ch != nil { + ch <- ev + } +} + +func opencodeGlobalEvent(eventType string, props map[string]any) map[string]any { + return map[string]any{ + "directory": "/tmp/iop-test", + "payload": map[string]any{ + "id": "evt_test", + "type": eventType, + "properties": props, + }, + } +} + +func opencodeSSEProfile(attachURL string, opts ...string) config.CLIProfileConf { + args := []string{"--attach", attachURL, "--title", "test"} + args = append(args, opts...) + return config.CLIProfileConf{ + Command: "/bin/true", + Args: args, + Mode: "opencode-sse", + } +} + +func TestCLIExecuteOpencodeSSE_StreamsTextDeltas(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_1") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL, "--model", "prov/m1", "--dangerously-skip-permissions"), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + done := make(chan error, 1) + go func() { + done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-1", + Target: "opencode", + Input: map[string]any{"prompt": "hello"}, + }, sink) + }() + + // Wait until prompt_async has been received, then push events. + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if fake.promptCalls.Load() == 0 { + t.Fatal("prompt_async was not called") + } + + fake.push(map[string]any{ + "type": "session.next.text.delta", + "properties": map[string]any{ + "sessionID": "ses_1", + "delta": "Hello ", + }, + }) + fake.push(map[string]any{ + "type": "session.next.text.delta", + "properties": map[string]any{ + "sessionID": "ses_1", + "delta": "world", + }, + }) + fake.push(map[string]any{ + "type": "session.next.step.ended", + "properties": map[string]any{ + "sessionID": "ses_1", + "tokens": map[string]any{"input": 5.0, "output": 2.0}, + }, + }) + fake.push(map[string]any{ + "type": "session.idle", + "properties": map[string]any{ + "sessionID": "ses_1", + }, + }) + + select { + case err := <-done: + if err != nil { + t.Fatalf("execute: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return") + } + + events := sink.Events() + if combined := testutil.CollectDeltas(events); combined != "Hello world" { + t.Fatalf("deltas: got %q, want %q", combined, "Hello world") + } + var complete *noderuntime.RuntimeEvent + for i := range events { + if events[i].Type == noderuntime.EventTypeComplete { + complete = &events[i] + } + } + if complete == nil { + t.Fatal("expected complete event") + } + if complete.Usage == nil || complete.Usage.InputTokens != 5 || complete.Usage.OutputTokens != 2 { + t.Fatalf("expected usage 5/2, got %+v", complete.Usage) + } + if fake.createCalls.Load() != 1 { + t.Errorf("expected 1 create session call, got %d", fake.createCalls.Load()) + } + + fake.mu.Lock() + body := append([]byte(nil), fake.lastPromptBody...) + fake.mu.Unlock() + var decoded struct { + Parts []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"parts"` + Model struct { + ProviderID string `json:"providerID"` + ModelID string `json:"modelID"` + } `json:"model"` + } + if err := json.Unmarshal(body, &decoded); err != nil { + t.Fatalf("decode prompt body: %v (body=%q)", err, string(body)) + } + if len(decoded.Parts) != 1 || decoded.Parts[0].Type != "text" || decoded.Parts[0].Text != "hello" { + t.Errorf("parts: got %+v", decoded.Parts) + } + if decoded.Model.ProviderID != "prov" || decoded.Model.ModelID != "m1" { + t.Errorf("model: got %+v", decoded.Model) + } +} + +func TestCLIExecuteOpencodeSSE_GlobalMessagePartDeltaStreamsTextOnly(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_global") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + done := make(chan error, 1) + go func() { + done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-global", + Target: "opencode", + Input: map[string]any{"prompt": "hello"}, + }, sink) + }() + + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if fake.promptCalls.Load() == 0 { + t.Fatal("prompt_async was not called") + } + + fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{ + "sessionID": "ses_global", + "part": map[string]any{"id": "prt_reason", "type": "reasoning", "text": ""}, + })) + fake.push(opencodeGlobalEvent("message.part.delta", map[string]any{ + "sessionID": "ses_global", + "partID": "prt_reason", + "field": "text", + "delta": "thinking", + })) + fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{ + "sessionID": "ses_global", + "part": map[string]any{"id": "prt_text", "type": "text", "text": ""}, + })) + fake.push(opencodeGlobalEvent("message.part.delta", map[string]any{ + "sessionID": "ses_global", + "partID": "prt_text", + "field": "text", + "delta": "OK", + })) + fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{ + "sessionID": "ses_global", + "part": map[string]any{ + "id": "prt_finish", + "type": "step-finish", + "tokens": map[string]any{"input": 7.0, "output": 1.0}, + }, + })) + fake.push(opencodeGlobalEvent("session.status", map[string]any{ + "sessionID": "ses_global", + "status": map[string]any{"type": "idle"}, + })) + + select { + case err := <-done: + if err != nil { + t.Fatalf("execute: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return") + } + + events := sink.Events() + if combined := testutil.CollectDeltas(events); combined != "OK" { + t.Fatalf("deltas: got %q, want %q", combined, "OK") + } + var complete *noderuntime.RuntimeEvent + for i := range events { + if events[i].Type == noderuntime.EventTypeComplete { + complete = &events[i] + } + } + if complete == nil { + t.Fatal("expected complete event") + } + if complete.Usage == nil || complete.Usage.InputTokens != 7 || complete.Usage.OutputTokens != 1 { + t.Fatalf("expected usage 7/1, got %+v", complete.Usage) + } +} + +func TestCLIExecuteOpencodeSSE_PromptAsyncSendsModelObject(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_m") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL, "--model", "ollama-dgx/qwen3.6:35b-a3b-bf16"), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + done := make(chan error, 1) + go func() { + done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-m", + Target: "opencode", + Input: map[string]any{"prompt": "hello"}, + }, sink) + }() + + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if fake.promptCalls.Load() == 0 { + t.Fatal("prompt_async was not called") + } + fake.push(map[string]any{ + "type": "session.idle", + "properties": map[string]any{"sessionID": "ses_m"}, + }) + select { + case err := <-done: + if err != nil { + t.Fatalf("execute: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return") + } + + fake.mu.Lock() + body := append([]byte(nil), fake.lastPromptBody...) + fake.mu.Unlock() + var decoded struct { + Model map[string]any `json:"model"` + } + if err := json.Unmarshal(body, &decoded); err != nil { + t.Fatalf("decode body: %v (body=%q)", err, string(body)) + } + if decoded.Model["providerID"] != "ollama-dgx" { + t.Errorf("providerID: got %v", decoded.Model["providerID"]) + } + if decoded.Model["modelID"] != "qwen3.6:35b-a3b-bf16" { + t.Errorf("modelID: got %v", decoded.Model["modelID"]) + } +} + +func TestCLIExecuteOpencodeSSE_InvalidModelEmitsError(t *testing.T) { + _, srv := newOpencodeFakeServer(t, "ses_bad") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL, "--model", "missing-slash"), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-bad", + Target: "opencode", + Input: map[string]any{"prompt": "hi"}, + }, sink) + if err == nil { + t.Fatal("expected error for invalid model, got nil") + } + if !strings.Contains(err.Error(), "missing provider/model separator") { + t.Fatalf("unexpected error: %v", err) + } + var sawErr bool + for _, e := range sink.Events() { + if e.Type == noderuntime.EventTypeError && strings.Contains(e.Error, "missing provider/model separator") { + sawErr = true + } + } + if !sawErr { + t.Fatal("expected error event with model parse failure") + } +} + +func TestCLIExecuteOpencodeSSE_SessionStatusObjectIdleCompletes(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_obj") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + done := make(chan error, 1) + go func() { + done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-obj", + Target: "opencode", + Input: map[string]any{"prompt": "hi"}, + }, sink) + }() + + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + + fake.push(map[string]any{ + "type": "session.next.text.delta", + "properties": map[string]any{ + "sessionID": "ses_obj", + "delta": "done", + }, + }) + fake.push(map[string]any{ + "type": "session.status", + "properties": map[string]any{ + "sessionID": "ses_obj", + "status": map[string]any{"type": "idle"}, + }, + }) + + select { + case err := <-done: + if err != nil { + t.Fatalf("execute: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return") + } + var sawComplete bool + for _, e := range sink.Events() { + if e.Type == noderuntime.EventTypeComplete { + sawComplete = true + } + } + if !sawComplete { + t.Fatal("expected complete event after object-idle session.status") + } +} + +func TestCLIExecuteOpencodeSSE_RequireExistingWithoutSessionErrors(t *testing.T) { + _, srv := newOpencodeFakeServer(t, "ses_1") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + err := c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-x", + Target: "opencode", + SessionID: "abs", + SessionMode: noderuntime.SessionModeRequireExisting, + Input: map[string]any{"prompt": "hi"}, + }, sink) + if err == nil { + t.Fatal("expected error when SessionModeRequireExisting and no cached session") + } + if !strings.Contains(err.Error(), "no persistent session") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestCLIExecuteOpencodeSSE_SessionErrorEmitsRuntimeError(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_1") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + done := make(chan error, 1) + go func() { + done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-err", + Target: "opencode", + Input: map[string]any{"prompt": "hi"}, + }, sink) + }() + + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + + fake.push(map[string]any{ + "type": "session.error", + "properties": map[string]any{ + "sessionID": "ses_1", + "error": map[string]any{ + "name": "ModelError", + "data": map[string]any{"message": "Model not found: foo/bar"}, + }, + }, + }) + + select { + case err := <-done: + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "Model not found: foo/bar") { + t.Fatalf("unexpected error: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return") + } + + var gotErr string + for _, e := range sink.Events() { + if e.Type == noderuntime.EventTypeError { + gotErr = e.Error + } + } + if gotErr != "Model not found: foo/bar" { + t.Fatalf("expected runtime error event, got %q", gotErr) + } +} + +func TestCLIExecuteOpencodeSSE_ContextCancelAbortsSession(t *testing.T) { + fake, srv := newOpencodeFakeServer(t, "ses_1") + + cfg := config.CLIConf{ + Enabled: true, + Profiles: map[string]config.CLIProfileConf{ + "opencode": opencodeSSEProfile(srv.URL), + }, + } + c := clipkg.New(cfg, zap.NewNop()) + sink := &testutil.FakeSink{} + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- c.Execute(ctx, noderuntime.ExecutionSpec{ + RunID: "run-cancel", + Target: "opencode", + Input: map[string]any{"prompt": "hi"}, + }, sink) + }() + + deadline := time.Now().Add(2 * time.Second) + for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + cancel() + + select { + case err := <-done: + if err != noderuntime.ErrRunCancelled { + t.Fatalf("expected ErrRunCancelled, got %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("execute did not return after cancel") + } + + var cancelEvent bool + var msg string + for _, e := range sink.Events() { + if e.Type == noderuntime.EventTypeCancelled { + cancelEvent = true + msg = e.Message + } + } + if !cancelEvent { + t.Fatal("expected cancelled event") + } + if msg != "user-cancel" { + t.Fatalf("expected user-cancel, got %q", msg) + } + + // Allow a short window for the abort POST to arrive. + deadline = time.Now().Add(1 * time.Second) + for fake.abortCalls.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if fake.abortCalls.Load() == 0 { + t.Fatal("expected /session/{id}/abort to be called") + } +} diff --git a/apps/node/internal/adapters/cli/opencode_sse_internal_test.go b/apps/node/internal/adapters/cli/opencode_sse_internal_test.go new file mode 100644 index 0000000..5146abc --- /dev/null +++ b/apps/node/internal/adapters/cli/opencode_sse_internal_test.go @@ -0,0 +1,202 @@ +package cli + +import ( + "encoding/json" + "testing" +) + +func TestParseOpencodeRunArgs(t *testing.T) { + args := []string{ + "--attach", "http://127.0.0.1:9876", + "--model", "ollama-dgx/qwen3.6:35b-a3b-bf16", + "--agent", "build", + "--variant", "fast", + "--title", "demo", + "--dir", "/work", + "--dangerously-skip-permissions", + } + opts := parseOpencodeRunArgs(args) + if opts.AttachURL != "http://127.0.0.1:9876" { + t.Errorf("attach: got %q", opts.AttachURL) + } + if opts.Model != "ollama-dgx/qwen3.6:35b-a3b-bf16" { + t.Errorf("model: got %q", opts.Model) + } + if opts.Agent != "build" { + t.Errorf("agent: got %q", opts.Agent) + } + if opts.Variant != "fast" { + t.Errorf("variant: got %q", opts.Variant) + } + if opts.Title != "demo" { + t.Errorf("title: got %q", opts.Title) + } + if opts.Dir != "/work" { + t.Errorf("dir: got %q", opts.Dir) + } + if !opts.DangerouslySkipPermissions { + t.Error("dangerously-skip-permissions should be true") + } +} + +func TestParseOpencodeRunArgs_NoAttach(t *testing.T) { + args := []string{"--title", "untitled", "--model", "m"} + opts := parseOpencodeRunArgs(args) + if opts.AttachURL != "" { + t.Errorf("expected empty attach, got %q", opts.AttachURL) + } + if opts.Title != "untitled" || opts.Model != "m" { + t.Errorf("unexpected: %+v", opts) + } +} + +func TestOpencodeModelPayload(t *testing.T) { + cases := []struct { + in string + wantProv string + wantMod string + wantErr bool + }{ + {"ollama-dgx/qwen3.6:35b-a3b-bf16", "ollama-dgx", "qwen3.6:35b-a3b-bf16", false}, + {"anthropic/claude-sonnet-4-6", "anthropic", "claude-sonnet-4-6", false}, + {"openrouter/google/gemini-2.5-pro", "openrouter", "google/gemini-2.5-pro", false}, + {"missing-slash", "", "", true}, + {"/missing-provider", "", "", true}, + {"missing-model/", "", "", true}, + } + for _, tc := range cases { + got, err := opencodeModelPayload(tc.in) + if tc.wantErr { + if err == nil { + t.Errorf("opencodeModelPayload(%q): expected error, got nil", tc.in) + } + continue + } + if err != nil { + t.Errorf("opencodeModelPayload(%q): unexpected error %v", tc.in, err) + continue + } + if got["providerID"] != tc.wantProv { + t.Errorf("opencodeModelPayload(%q) providerID = %v, want %q", tc.in, got["providerID"], tc.wantProv) + } + if got["modelID"] != tc.wantMod { + t.Errorf("opencodeModelPayload(%q) modelID = %v, want %q", tc.in, got["modelID"], tc.wantMod) + } + } +} + +func TestOpencodeStatusIdle(t *testing.T) { + cases := []struct { + name string + props map[string]any + want bool + }{ + {"string idle", map[string]any{"status": "idle"}, true}, + {"object idle", map[string]any{"status": map[string]any{"type": "idle"}}, true}, + {"object busy", map[string]any{"status": map[string]any{"type": "busy"}}, false}, + {"string busy", map[string]any{"status": "busy"}, false}, + {"missing", map[string]any{}, false}, + {"non-map number", map[string]any{"status": 1.0}, false}, + {"nil props", nil, false}, + } + for _, tc := range cases { + if got := opencodeStatusIdle(tc.props); got != tc.want { + t.Errorf("opencodeStatusIdle(%s) = %v, want %v", tc.name, got, tc.want) + } + } +} + +func TestOpencodeEventEnvelope(t *testing.T) { + cases := []struct { + name string + in string + want string + ok bool + }{ + { + name: "plain event", + in: `{"id":"evt_1","type":"session.idle","properties":{"sessionID":"ses_1"}}`, + want: "session.idle", + ok: true, + }, + { + name: "global payload event", + in: `{"directory":"/work","payload":{"id":"evt_1","type":"message.part.delta","properties":{"sessionID":"ses_1"}}}`, + want: "message.part.delta", + ok: true, + }, + { + name: "sync event ignored", + in: `{"directory":"/work","payload":{"type":"sync","syncEvent":{"type":"message.part.delta.1"}}}`, + ok: false, + }, + } + for _, tc := range cases { + got, ok := opencodeEventEnvelope([]byte(tc.in)) + if ok != tc.ok { + t.Fatalf("%s: ok=%v, want %v", tc.name, ok, tc.ok) + } + if ok && got.Type != tc.want { + t.Fatalf("%s: type=%q, want %q", tc.name, got.Type, tc.want) + } + } +} + +func TestOpencodeMessagePartDeltaText(t *testing.T) { + partTypes := map[string]string{ + "prt_text": "text", + "prt_reasoning": "reasoning", + } + cases := []struct { + name string + props map[string]any + want string + }{ + {"text part", map[string]any{"partID": "prt_text", "field": "text", "delta": "OK"}, "OK"}, + {"reasoning part ignored", map[string]any{"partID": "prt_reasoning", "field": "text", "delta": "thinking"}, ""}, + {"unknown part ignored", map[string]any{"partID": "prt_missing", "field": "text", "delta": "lost"}, ""}, + {"non text field ignored", map[string]any{"partID": "prt_text", "field": "metadata", "delta": "lost"}, ""}, + } + for _, tc := range cases { + if got := opencodeMessagePartDeltaText(tc.props, partTypes); got != tc.want { + t.Fatalf("%s: got %q, want %q", tc.name, got, tc.want) + } + } +} + +func TestOpencodeStepTokensFromPart(t *testing.T) { + var props map[string]any + if err := json.Unmarshal([]byte(`{"part":{"type":"step-finish","tokens":{"input":7,"output":1}}}`), &props); err != nil { + t.Fatal(err) + } + in, out, ok := opencodeStepTokens(props) + if !ok || in != 7 || out != 1 { + t.Fatalf("tokens: got %d/%d ok=%v, want 7/1 true", in, out, ok) + } +} + +func TestDecodeSSEDataLine(t *testing.T) { + cases := []struct { + in string + want string + shouldGet bool + }{ + {"data: {\"a\":1}", `{"a":1}`, true}, + {"data:{\"a\":1}", `{"a":1}`, true}, + {"event: ping", "", false}, + {": comment", "", false}, + {"", "", false}, + {"data:", "", false}, + {"data: ", "", false}, + } + for _, tc := range cases { + got, ok := decodeSSEDataLine([]byte(tc.in)) + if ok != tc.shouldGet { + t.Errorf("decodeSSEDataLine(%q) ok=%v, want %v", tc.in, ok, tc.shouldGet) + continue + } + if ok && string(got) != tc.want { + t.Errorf("decodeSSEDataLine(%q) = %q, want %q", tc.in, string(got), tc.want) + } + } +} diff --git a/apps/node/internal/adapters/factory_internal_test.go b/apps/node/internal/adapters/factory_internal_test.go index 7e17741..00d4fee 100644 --- a/apps/node/internal/adapters/factory_internal_test.go +++ b/apps/node/internal/adapters/factory_internal_test.go @@ -160,6 +160,52 @@ func TestCLIConfFromProto_NilSettings(t *testing.T) { } } +func TestCLIConfFromProto_OpencodeSSEMode(t *testing.T) { + m := &iop.CLIAdapterConfig{ + Profiles: map[string]*iop.CLIProfileConfig{ + "opencode": { + Command: "/usr/local/bin/opencode", + Args: []string{"--model", "m1", "--dangerously-skip-permissions"}, + Mode: "opencode-sse", + }, + }, + } + cfg := cliConfFromProto(m) + prof, ok := cfg.Profiles["opencode"] + if !ok { + t.Fatal("expected opencode profile") + } + if prof.Mode != "opencode-sse" { + t.Errorf("mode: got %q", prof.Mode) + } + if len(prof.Args) != 3 { + t.Fatalf("args: got %#v", prof.Args) + } +} + +func TestCLIConfFromStruct_OpencodeSSEModeLegacy(t *testing.T) { + st, err := structpb.NewStruct(map[string]any{ + "profiles": map[string]any{ + "opencode": map[string]any{ + "command": "/usr/local/bin/opencode", + "args": []any{"--model", "m1"}, + "mode": "opencode-sse", + }, + }, + }) + if err != nil { + t.Fatalf("structpb: %v", err) + } + cfg := cliConfFromStruct(st) + prof, ok := cfg.Profiles["opencode"] + if !ok { + t.Fatal("expected opencode profile") + } + if prof.Mode != "opencode-sse" { + t.Errorf("mode: got %q", prof.Mode) + } +} + // Legacy structpb fallback regression tests — these exercise the path taken // when an older edge sends AdapterConfig.settings instead of the typed oneof. diff --git a/bin/node.sh b/bin/node.sh index f285056..3fa8153 100755 --- a/bin/node.sh +++ b/bin/node.sh @@ -7,4 +7,30 @@ CONFIG_FILE="$REPO_ROOT/configs/node.yaml" cd "$REPO_ROOT" echo "[node] config=$CONFIG_FILE" + +EDGE_ADDR="${IOP_EDGE_ADDR:-$(sed -nE 's/^[[:space:]]*edge_addr:[[:space:]]*"?([^"#]+)"?[[:space:]]*(#.*)?$/\1/p' "$CONFIG_FILE" | head -n1)}" +EDGE_ADDR="${EDGE_ADDR//[[:space:]]/}" +EDGE_HOST="${EDGE_ADDR%:*}" +EDGE_PORT="${EDGE_ADDR##*:}" +EDGE_HOST="${EDGE_HOST#[}" +EDGE_HOST="${EDGE_HOST%]}" +WAIT_TIMEOUT="${IOP_NODE_WAIT_TIMEOUT:-30}" +WAIT_INTERVAL="${IOP_NODE_WAIT_INTERVAL:-1}" + +if [[ -z "$EDGE_HOST" || -z "$EDGE_PORT" || "$EDGE_HOST" == "$EDGE_PORT" ]]; then + echo "[node] invalid edge_addr in $CONFIG_FILE: ${EDGE_ADDR:-}" >&2 + exit 1 +fi + +echo "[node] waiting for edge at $EDGE_HOST:$EDGE_PORT timeout=${WAIT_TIMEOUT}s" +deadline=$((SECONDS + WAIT_TIMEOUT)) +while ! timeout 1 bash -c 'cat < /dev/null > /dev/tcp/"$1"/"$2"' _ "$EDGE_HOST" "$EDGE_PORT" 2>/dev/null; do + if (( SECONDS >= deadline )); then + echo "[node] edge is not reachable at $EDGE_HOST:$EDGE_PORT; start ./bin/edge.sh first or update configs/node.yaml transport.edge_addr" >&2 + exit 1 + fi + sleep "$WAIT_INTERVAL" +done +echo "[node] edge is reachable" + exec go run ./apps/node/cmd/node serve --config "$CONFIG_FILE" diff --git a/configs/edge.yaml b/configs/edge.yaml index 5115860..835acb2 100644 --- a/configs/edge.yaml +++ b/configs/edge.yaml @@ -17,7 +17,7 @@ metrics: console: adapter: "cli" - agent: "cline-dgx" + target: "claude" session_id: "default" background: false timeout_sec: 300 @@ -85,18 +85,15 @@ nodes: opencode: command: "/config/.npm-global/bin/opencode" args: - - "run" - "--title" - "untitle" - "--model" - "ollama-dgx/qwen3.6:35b-a3b-bf16" - - "--format" - - "json" - "--dangerously-skip-permissions" env: [] persistent: false terminal: false - output_format: "opencode-json" + mode: "opencode-sse" cline-dgx: command: "/config/.npm-global/bin/cline" args: diff --git a/packages/config/config.go b/packages/config/config.go index 61b5d6f..da9251e 100644 --- a/packages/config/config.go +++ b/packages/config/config.go @@ -47,20 +47,29 @@ type EdgeServerConf struct { type EdgeConsoleConf struct { Adapter string `mapstructure:"adapter" yaml:"adapter"` - Agent string `mapstructure:"agent" yaml:"agent"` - Model string `mapstructure:"model" yaml:"model"` + Target string `mapstructure:"target" yaml:"target"` + Agent string `mapstructure:"agent" yaml:"agent"` // legacy alias for target + Model string `mapstructure:"model" yaml:"model"` // legacy alias for target SessionID string `mapstructure:"session_id" yaml:"session_id"` Background bool `mapstructure:"background" yaml:"background"` TimeoutSec int `mapstructure:"timeout_sec" yaml:"timeout_sec"` } -func (c EdgeConsoleConf) ResolveAgent() string { +func (c EdgeConsoleConf) ResolveTarget() string { + if c.Target != "" { + return c.Target + } if c.Agent != "" { return c.Agent } return c.Model } +// ResolveAgent is kept for callers still using the legacy console.agent name. +func (c EdgeConsoleConf) ResolveAgent() string { + return c.ResolveTarget() +} + type TransportConf struct { EdgeAddr string `mapstructure:"edge_addr" yaml:"edge_addr"` Token string `mapstructure:"token" yaml:"token"` @@ -160,8 +169,12 @@ func LoadEdge(cfgFile string) (*EdgeConfig, error) { if err := v.Unmarshal(&cfg); err != nil { return nil, err } - if !v.InConfig("console.agent") && v.InConfig("console.model") { - cfg.Console.Agent = cfg.Console.Model + if !v.InConfig("console.target") { + if v.InConfig("console.agent") { + cfg.Console.Target = cfg.Console.Agent + } else if v.InConfig("console.model") { + cfg.Console.Target = cfg.Console.Model + } } return &cfg, nil } @@ -178,7 +191,7 @@ func setEdgeDefaults(v *viper.Viper) { v.SetDefault("metrics.port", 9092) v.SetDefault("tls.enabled", false) v.SetDefault("console.adapter", "cli") - v.SetDefault("console.agent", "claude") + v.SetDefault("console.target", "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 81e1e64..80a7823 100644 --- a/packages/config/config_test.go +++ b/packages/config/config_test.go @@ -60,8 +60,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.ResolveAgent() != "claude" { - t.Fatalf("expected default console.agent=%q, got %q", "claude", cfg.Console.ResolveAgent()) + if cfg.Console.ResolveTarget() != "claude" { + t.Fatalf("expected default console.target=%q, got %q", "claude", cfg.Console.ResolveTarget()) } } @@ -119,7 +119,41 @@ func TestLoadEdge_ConsoleTimeoutOverride(t *testing.T) { } } -func TestLoadEdge_ConsoleAgentFallbackFromLegacyModel(t *testing.T) { +func TestLoadEdge_ConsoleTargetOverride(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + yaml := "server:\n listen: \"0.0.0.0:9090\"\nconsole:\n target: \"codex\"\n" + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + + cfg, err := config.LoadEdge(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.Console.ResolveTarget() != "codex" { + t.Fatalf("expected console.target to resolve to %q, got %q", "codex", cfg.Console.ResolveTarget()) + } +} + +func TestLoadEdge_ConsoleTargetFallbackFromLegacyAgent(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + yaml := "server:\n listen: \"0.0.0.0:9090\"\nconsole:\n agent: \"codex\"\n" + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + + cfg, err := config.LoadEdge(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.Console.ResolveTarget() != "codex" { + t.Fatalf("expected legacy console.agent fallback to resolve to %q, got %q", "codex", cfg.Console.ResolveTarget()) + } +} + +func TestLoadEdge_ConsoleTargetFallbackFromLegacyModel(t *testing.T) { dir := t.TempDir() f := filepath.Join(dir, "edge.yaml") yaml := "server:\n listen: \"0.0.0.0:9090\"\nconsole:\n model: \"codex\"\n" @@ -131,8 +165,8 @@ func TestLoadEdge_ConsoleAgentFallbackFromLegacyModel(t *testing.T) { if err != nil { t.Fatalf("load: %v", err) } - if cfg.Console.ResolveAgent() != "codex" { - t.Fatalf("expected legacy console.model fallback to resolve to %q, got %q", "codex", cfg.Console.ResolveAgent()) + if cfg.Console.ResolveTarget() != "codex" { + t.Fatalf("expected legacy console.model fallback to resolve to %q, got %q", "codex", cfg.Console.ResolveTarget()) } } @@ -372,6 +406,56 @@ nodes: } } +func TestLoadEdge_OpencodeSSEProfile(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + yaml := ` +server: + listen: "0.0.0.0:9090" +nodes: + - alias: "local" + adapters: + cli: + enabled: true + profiles: + opencode: + command: "/usr/local/bin/opencode" + args: + - "--title" + - "untitle" + - "--model" + - "ollama-dgx/qwen3.6:35b-a3b-bf16" + - "--dangerously-skip-permissions" + mode: "opencode-sse" +` + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + cfg, err := config.LoadEdge(f) + if err != nil { + t.Fatalf("load: %v", err) + } + if len(cfg.Nodes) == 0 { + t.Fatal("expected 1 node") + } + prof, ok := cfg.Nodes[0].Adapters.CLI.Profiles["opencode"] + if !ok { + t.Fatal("expected opencode profile") + } + if prof.Mode != "opencode-sse" { + t.Errorf("mode: got %q", prof.Mode) + } + expectedArgs := []string{"--title", "untitle", "--model", "ollama-dgx/qwen3.6:35b-a3b-bf16", "--dangerously-skip-permissions"} + if len(prof.Args) != len(expectedArgs) { + t.Fatalf("args len: got %d, want %d", len(prof.Args), len(expectedArgs)) + } + for i, want := range expectedArgs { + if prof.Args[i] != want { + t.Errorf("args[%d]: got %q, want %q", i, prof.Args[i], want) + } + } +} + func TestCLIProfileConf_CompletionMarkerEmpty(t *testing.T) { dir := t.TempDir() f := filepath.Join(dir, "edge.yaml")