From 898da6e3afcb46bc8b78bf1130e0777e0f3cacef Mon Sep 17 00:00:00 2001 From: toki Date: Mon, 15 Jun 2026 14:12:40 +0900 Subject: [PATCH] feat(node): openai compatible adapter implementation and test fixes - Add openai_compat adapter package - Move 02+01 OpenAI compatible adapter files to archive - Fix adapter factory and test files - Update blackbox tests --- .../code_review_local_G08_0.log} | 92 +++- .../complete.log | 41 ++ .../plan_local_G08_0.log} | 0 .../adapters/adapters_blackbox_test.go | 124 +++++ apps/node/internal/adapters/factory.go | 57 +++ .../adapters/factory_internal_test.go | 52 +++ .../adapters/openai_compat/openai_compat.go | 438 ++++++++++++++++++ .../openai_compat/openai_compat_test.go | 330 +++++++++++++ 8 files changed, 1109 insertions(+), 25 deletions(-) rename agent-task/{m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/CODE_REVIEW-local-G08.md => archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log} (57%) create mode 100644 agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/complete.log rename agent-task/{m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/PLAN-local-G08.md => archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/plan_local_G08_0.log} (100%) create mode 100644 apps/node/internal/adapters/openai_compat/openai_compat.go create mode 100644 apps/node/internal/adapters/openai_compat/openai_compat_test.go diff --git a/agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/CODE_REVIEW-local-G08.md b/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log similarity index 57% rename from agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/CODE_REVIEW-local-G08.md rename to agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log index 24c91c5..04b775c 100644 --- a/agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/CODE_REVIEW-local-G08.md +++ b/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log @@ -42,44 +42,52 @@ task=m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter, pla | 항목 | 완료 여부 | |------|---------| -| [API-1] Common OpenAI-compatible adapter | [ ] | -| [API-2] Factory registration | [ ] | -| [API-3] Node execution regression coverage | [ ] | +| [API-1] Common OpenAI-compatible adapter | [x] | +| [API-2] Factory registration | [x] | +| [API-3] Node execution regression coverage | [x] | ## 구현 체크리스트 -- [ ] 구현 시작 전 `01_config_contract`의 `complete.log`를 확인하고, 선행 계약의 type/field 이름을 이 plan에 맞춰 반영한다. -- [ ] `openai_compat` Node adapter를 추가하고 `/v1/models`, `/v1/chat/completions` streaming, headers, option passthrough를 구현한다. -- [ ] factory가 typed/legacy `openai_compat` payload를 registry에 등록하도록 연결한다. -- [ ] Node adapter/factory/blackbox 테스트를 작성하고 기존 vLLM 테스트가 그대로 PASS하는지 확인한다. -- [ ] 계획의 중간 검증과 최종 검증 명령을 실행하고 실제 출력을 review stub에 붙인다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] 구현 시작 전 `01_config_contract`의 `complete.log`를 확인하고, 선행 계약의 type/field 이름을 이 plan에 맞춰 반영한다. +- [x] `openai_compat` Node adapter를 추가하고 `/v1/models`, `/v1/chat/completions` streaming, headers, option passthrough를 구현한다. +- [x] factory가 typed/legacy `openai_compat` payload를 registry에 등록하도록 연결한다. +- [x] Node adapter/factory/blackbox 테스트를 작성하고 기존 vLLM 테스트가 그대로 PASS하는지 확인한다. +- [x] 계획의 중간 검증과 최종 검증 명령을 실행하고 실제 출력을 review stub에 붙인다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. ## 코드리뷰 전용 체크리스트 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. > 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G08_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_local_G08_M.log`로 아카이브한다. -- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. -- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리 `agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/`를 `agent-task/archive/YYYY/MM/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. -- [ ] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. -- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-lemonade-provider-serving-validation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G08_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G08_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/`를 `agent-task/archive/YYYY/MM/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-lemonade-provider-serving-validation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. - [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-local-G08.md`와 `CODE_REVIEW-local-G08.md`를 작성하고 `complete.log`를 작성하지 않는다. - [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. - [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- 선행 `01_config_contract`의 active `complete.log`는 이미 archive 이동되어 active 디렉터리에 없었다. 권한상 `agent-task/archive/**` 직접 열람이 막혀 있어, 선행 계약 충족 여부를 `01_config_contract`의 실제 산출물(`config.OpenAICompatConf`/`OpenAICompatInstanceConf`, proto `OpenAICompatAdapterConfig`, generated `AdapterConfig.GetOpenaiCompat()`)이 코드에 존재하고 커밋 `76a8ee5`("lemonade provider serving validation - config contract ...")에 반영된 것으로 확인했다. type/field 이름은 선행 계약 그대로 사용했고 별도 rename은 없었다. +- 계획은 vLLM 공통 helper 추출을 "필요 시"로 허용했으나, vLLM adapter name/behavior/test 회귀 위험을 0으로 두기 위해 `openai_compat` 패키지를 자족적으로 구현하고 `apps/node/internal/adapters/vllm/*`는 수정하지 않았다. 결과적으로 `joinURL` 대신 `/v1` 중복 방지를 처리하는 `joinOpenAIPath`를 openai_compat 내부에만 추가했다. +- 계획 외 추가 테스트로 `TestOpenAICompatJoinOpenAIPathNoDuplicateV1`(endpoint가 root 또는 trailing `/v1`일 때 경로 중복 방지)와 `TestBuildFromPayload_OpenAICompatTypedRegistered`(typed payload registry key 검증)를 더했다. 계획 명시 테스트는 모두 포함했다. +- 검증 명령은 계획 그대로 실행했다(대체 없음). ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- **Content-Type 소유권**: POST `/v1/chat/completions`에서 adapter가 항상 `Content-Type: application/json`을 소유한다. user header에 `Content-Type`이 있어도 무시(override 차단)하고, 그 외 auth/secret header(`Authorization` 등)는 그대로 주입한다. `applyHeaders(req, jsonBody)`로 GET `/v1/models`(auth header만 주입)와 POST(content-type 소유)를 한 곳에서 처리한다. (`TestOpenAICompatExecuteSendsHeaders`로 검증) +- **option passthrough 우선순위**: request body를 `map[string]any`로 만들고 `input["options"]`를 top-level OpenAI-compatible field로 먼저 복사한 뒤 adapter-owned `model`/`messages`/`stream=true`를 나중에 써서 항상 우선하게 했다. options가 `model`/`stream`을 담아도 덮어쓰지 못한다. `tools`/`format`/`think`/`keep_alive`는 input에 있을 때만 복사한다. (`TestOpenAICompatExecutePassesOptionsAsTopLevelFields`로 검증) +- **streaming/usage/finish_reason**: SSE `data: [DONE]`에서 complete를 emit한다. `delta.content`→`EventTypeDelta`, `delta.reasoning_content`→`EventTypeReasoningDelta`로 분리한다. chunk `usage`가 있으면 `RuntimeEvent.Usage`(input/output tokens)에 반영하고, 없으면 vLLM과 동일하게 output token field-count fallback을 쓴다. `finish_reason`은 complete event `Metadata["finish_reason"]`로 보존한다. +- **provider status baseline**: `Capabilities`/`ProbeProvider` 모두 `/v1/models`의 `data[].id`를 사용하고 `runtime.NormalizeProviderStatus`로 `available`/`unavailable`만 반환한다. probe 실패 시 `unavailable`, baseline 밖 상태로 확장하지 않는다. provider label은 `ProviderProbeResult.Metadata["provider"]`로만 노출한다(공통 상태 모델 불변). +- **endpoint path 정규화**: `joinOpenAIPath`는 endpoint path가 이미 `/v1`로 끝나면 path의 선행 `/v1`을 제거해 `/v1/v1/...` 중복을 방지한다. root endpoint와 trailing-`/v1` endpoint를 모두 지원한다. +- **legacy fallback**: factory는 typed `ac.GetOpenaiCompat()`가 있으면 `openAICompatConfFromProto`, 없으면 `openAICompatConfFromStruct`로 legacy `settings`(provider/endpoint/headers/queue fields)를 읽는다. headers는 structpb의 `map[string]any`에서 string 값만 추려 `map[string]string`으로 변환한다. ## 사용자 리뷰 요청 @@ -115,27 +123,44 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 ### API-1 중간 검증 ```bash $ go test -count=1 ./apps/node/internal/adapters/openai_compat ./apps/node/internal/adapters/vllm -(output) +ok iop/apps/node/internal/adapters/openai_compat 0.007s +ok iop/apps/node/internal/adapters/vllm 0.006s ``` ### API-2 중간 검증 ```bash $ go test -count=1 ./apps/node/internal/adapters -(output) +ok iop/apps/node/internal/adapters 0.008s ``` ### API-3 중간 검증 ```bash $ go test -count=1 ./apps/node/internal/adapters -(output) +ok iop/apps/node/internal/adapters 0.008s ``` ### 최종 검증 ```bash $ go test -count=1 ./apps/node/internal/adapters/openai_compat ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters ./apps/node/... -(output) +ok iop/apps/node/internal/adapters/openai_compat 0.007s +ok iop/apps/node/internal/adapters/vllm 0.010s +ok iop/apps/node/internal/adapters 0.010s +ok iop/apps/node/cmd/node 0.009s +ok iop/apps/node/internal/adapters/cli 46.684s +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status 39.779s +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama 0.008s +ok iop/apps/node/internal/bootstrap 0.262s +ok iop/apps/node/internal/node 2.682s +ok iop/apps/node/internal/router 0.006s +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store 0.036s +ok iop/apps/node/internal/terminal 0.554s +ok iop/apps/node/internal/transport 5.137s + $ git diff --check -(output) +(출력 없음, exit 0 — whitespace error 없음) ``` --- @@ -143,3 +168,20 @@ $ git diff --check > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** > If anything is blank, go back and fill it in before saving this file. > Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 검증 확인: + - `go test -count=1 ./apps/node/internal/adapters/openai_compat ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters ./apps/node/...` PASS + - `git diff --check` PASS +- 다음 단계: PASS 처리로 active plan/review를 로그로 아카이브하고 `complete.log` 작성 후 task 디렉터리를 `agent-task/archive/YYYY/MM/` 아래로 이동한다. diff --git a/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/complete.log b/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/complete.log new file mode 100644 index 0000000..3486d71 --- /dev/null +++ b/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/complete.log @@ -0,0 +1,41 @@ +# Complete - m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter + +## 완료 일시 + +2026-06-15 + +## 요약 + +OpenAI-compatible Node adapter implementation review completed in 1 loop with final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/plan_local_G08_0.log` | `agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log` | PASS | OpenAI-compatible adapter, factory registration, header/option passthrough, provider probe, and Node adapter tests matched the plan. | + +## 구현/정리 내용 + +- Added `openai_compat` Node adapter with `/v1/models` capabilities/probe, `/v1/chat/completions` SSE execution, configured headers, option passthrough, finish reason metadata, and usage fallback. +- Registered typed and legacy `openai_compat` payloads in the Node adapter factory. +- Added focused adapter, factory, and blackbox tests while preserving existing vLLM behavior. + +## 최종 검증 + +- `go test -count=1 ./apps/node/internal/adapters/openai_compat ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters ./apps/node/...` - PASS; changed adapter packages and full Node package tests completed successfully. +- `git diff --check` - PASS; no whitespace errors. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/lemonade-provider-serving-validation.md` +- Completed task ids: + - `availability`: PASS; evidence=`agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/plan_local_G08_0.log`, `agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/code_review_local_G08_0.log`; verification=`go test -count=1 ./apps/node/internal/adapters/openai_compat ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters ./apps/node/... && git diff --check` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/PLAN-local-G08.md b/agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/plan_local_G08_0.log similarity index 100% rename from agent-task/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/PLAN-local-G08.md rename to agent-task/archive/2026/06/m-lemonade-provider-serving-validation/02+01_openai_compatible_adapter/plan_local_G08_0.log diff --git a/apps/node/internal/adapters/adapters_blackbox_test.go b/apps/node/internal/adapters/adapters_blackbox_test.go index f073b57..a5d210f 100644 --- a/apps/node/internal/adapters/adapters_blackbox_test.go +++ b/apps/node/internal/adapters/adapters_blackbox_test.go @@ -583,3 +583,127 @@ func TestBuildFromPayload_VllmExecution(t *testing.T) { t.Errorf("InstanceKey: got %q want vllm@gpu", caps.InstanceKey) } } + +func TestBuildFromPayload_OpenAICompatTypedRegistered(t *testing.T) { + reg, err := adapters.BuildFromPayload(&iop.NodeConfigPayload{ + Adapters: []*iop.AdapterConfig{ + { + Type: "openai_compat", + Name: "lemonade-local", + Enabled: true, + Config: &iop.AdapterConfig_OpenaiCompat{OpenaiCompat: &iop.OpenAICompatAdapterConfig{ + Provider: "lemonade", + Endpoint: "http://localhost:8000", + }}, + }, + }, + }, zap.NewNop()) + if err != nil { + t.Fatalf("build: %v", err) + } + a, ok := reg.Get("lemonade-local") + if !ok { + t.Fatal("expected lemonade-local registered") + } + if a.Name() != "openai_compat" { + t.Fatalf("adapter name: got %q want openai_compat", a.Name()) + } + caps, _ := a.Capabilities(context.Background()) + if caps.InstanceKey != "lemonade-local" { + t.Errorf("InstanceKey: got %q want lemonade-local", caps.InstanceKey) + } +} + +func TestBuildFromPayload_OpenAICompatExecution(t *testing.T) { + var gotAuth string + var gotBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/models" { + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"lemonade-model"}]}`)) + return + } + if r.URL.Path != "/v1/chat/completions" { + t.Errorf("unexpected path: %s", r.URL.Path) + return + } + gotAuth = r.Header.Get("Authorization") + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Errorf("decode: %v", err) + return + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"LEMONADE_"}}]}`) + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"OK"},"finish_reason":"stop"}]}`) + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + })) + defer server.Close() + + reg, err := adapters.BuildFromPayload(&iop.NodeConfigPayload{ + Adapters: []*iop.AdapterConfig{ + { + Type: "openai_compat", + Name: "lemonade-local", + Enabled: true, + Config: &iop.AdapterConfig_OpenaiCompat{OpenaiCompat: &iop.OpenAICompatAdapterConfig{ + Provider: "lemonade", + Endpoint: server.URL, + Headers: map[string]string{"Authorization": "Bearer secret"}, + }}, + }, + }, + }, zap.NewNop()) + if err != nil { + t.Fatalf("build: %v", err) + } + + a, ok := reg.Get("lemonade-local") + if !ok { + t.Fatal("expected lemonade-local") + } + + sink := &testSink{} + if err := a.Execute(context.Background(), noderuntime.ExecutionSpec{ + RunID: "run-lemonade", + Target: "lemonade-model", + Input: map[string]any{ + "messages": []any{map[string]any{"role": "user", "content": "hi"}}, + "options": map[string]any{"temperature": 0.2, "max_tokens": float64(8)}, + }, + }, sink); err != nil { + t.Fatalf("Execute: %v", err) + } + + if gotAuth != "Bearer secret" { + t.Errorf("Authorization: got %q want Bearer secret", gotAuth) + } + if gotBody["model"] != "lemonade-model" { + t.Errorf("model: got %v want lemonade-model", gotBody["model"]) + } + if gotBody["stream"] != true { + t.Errorf("expected stream=true, got %v", gotBody["stream"]) + } + if gotBody["temperature"] != 0.2 || gotBody["max_tokens"] != float64(8) { + t.Errorf("options passthrough: got temperature=%v max_tokens=%v", gotBody["temperature"], gotBody["max_tokens"]) + } + + sink.mu.Lock() + defer sink.mu.Unlock() + var text string + var complete *noderuntime.RuntimeEvent + for i := range sink.events { + ev := sink.events[i] + if ev.Type == noderuntime.EventTypeDelta { + text += ev.Delta + } + if ev.Type == noderuntime.EventTypeComplete { + complete = &sink.events[i] + } + } + if text != "LEMONADE_OK" { + t.Errorf("streamed text: got %q want LEMONADE_OK", text) + } + if complete == nil || complete.Metadata["finish_reason"] != "stop" { + t.Errorf("expected complete with finish_reason stop, got %+v", complete) + } +} diff --git a/apps/node/internal/adapters/factory.go b/apps/node/internal/adapters/factory.go index 4a43483..b9104b5 100644 --- a/apps/node/internal/adapters/factory.go +++ b/apps/node/internal/adapters/factory.go @@ -9,6 +9,7 @@ import ( "iop/apps/node/internal/adapters/cli" "iop/apps/node/internal/adapters/mock" "iop/apps/node/internal/adapters/ollama" + "iop/apps/node/internal/adapters/openai_compat" "iop/apps/node/internal/adapters/vllm" "iop/packages/go/config" iop "iop/proto/gen/iop" @@ -43,6 +44,14 @@ func BuildFromPayload(payload *iop.NodeConfigPayload, logger *zap.Logger) (*Regi cfg = vllmConfFromStruct(ac.GetSettings()) } reg.RegisterKeyed(instanceKey, typeName, vllm.New(cfg, logger, instanceKey)) + case "openai_compat": + var cfg config.OpenAICompatConf + if m := ac.GetOpenaiCompat(); m != nil { + cfg = openAICompatConfFromProto(m) + } else { + cfg = openAICompatConfFromStruct(ac.GetSettings()) + } + reg.RegisterKeyed(instanceKey, typeName, openai_compat.New(cfg, logger, instanceKey)) case "cli": var cfg config.CLIConf if m := ac.GetCli(); m != nil { @@ -81,6 +90,19 @@ func vllmConfFromProto(m *iop.VllmAdapterConfig) config.VllmConf { } } +func openAICompatConfFromProto(m *iop.OpenAICompatAdapterConfig) config.OpenAICompatConf { + return config.OpenAICompatConf{ + Enabled: true, + Provider: m.GetProvider(), + Endpoint: m.GetEndpoint(), + Headers: m.GetHeaders(), + Capacity: int(m.GetCapacity()), + MaxQueue: int(m.GetMaxQueue()), + QueueTimeoutMS: int(m.GetQueueTimeoutMs()), + RequestTimeoutMS: int(m.GetRequestTimeoutMs()), + } +} + func cliConfFromProto(m *iop.CLIAdapterConfig) config.CLIConf { cfg := config.CLIConf{Enabled: true, Profiles: make(map[string]config.CLIProfileConf)} for name, p := range m.GetProfiles() { @@ -189,6 +211,41 @@ func vllmConfFromStruct(s *structpb.Struct) config.VllmConf { return cfg } +func openAICompatConfFromStruct(s *structpb.Struct) config.OpenAICompatConf { + cfg := config.OpenAICompatConf{Enabled: true} + if s == nil { + return cfg + } + m := s.AsMap() + if v, ok := m["provider"].(string); ok { + cfg.Provider = v + } + if v, ok := m["endpoint"].(string); ok { + cfg.Endpoint = v + } + if headers, ok := m["headers"].(map[string]any); ok { + cfg.Headers = make(map[string]string, len(headers)) + for k, hv := range headers { + if s, ok := hv.(string); ok { + cfg.Headers[k] = s + } + } + } + if v, ok := intSetting(m, "capacity"); ok { + cfg.Capacity = v + } + if v, ok := intSetting(m, "max_queue"); ok { + cfg.MaxQueue = v + } + if v, ok := intSetting(m, "queue_timeout_ms"); ok { + cfg.QueueTimeoutMS = v + } + if v, ok := intSetting(m, "request_timeout_ms"); ok { + cfg.RequestTimeoutMS = v + } + return cfg +} + func cliConfFromStruct(s *structpb.Struct) config.CLIConf { cfg := config.CLIConf{Enabled: true, Profiles: make(map[string]config.CLIProfileConf)} if s == nil { diff --git a/apps/node/internal/adapters/factory_internal_test.go b/apps/node/internal/adapters/factory_internal_test.go index 1116433..b106691 100644 --- a/apps/node/internal/adapters/factory_internal_test.go +++ b/apps/node/internal/adapters/factory_internal_test.go @@ -259,6 +259,58 @@ func TestVllmConfFromStruct_LegacyPayload(t *testing.T) { } } +func TestOpenAICompatConfFromProto(t *testing.T) { + cfg := openAICompatConfFromProto(&iop.OpenAICompatAdapterConfig{ + Provider: "lemonade", + Endpoint: "http://typed:8000", + Headers: map[string]string{"Authorization": "Bearer t"}, + Capacity: 4, + MaxQueue: 10, + QueueTimeoutMs: 2000, + RequestTimeoutMs: 45000, + }) + if !cfg.Enabled { + t.Fatal("expected enabled=true") + } + if cfg.Provider != "lemonade" || cfg.Endpoint != "http://typed:8000" { + t.Fatalf("config: got %+v", cfg) + } + if cfg.Headers["Authorization"] != "Bearer t" { + t.Fatalf("headers: got %+v", cfg.Headers) + } + if cfg.Capacity != 4 || cfg.MaxQueue != 10 || cfg.QueueTimeoutMS != 2000 || cfg.RequestTimeoutMS != 45000 { + t.Fatalf("queue config: got %+v", cfg) + } +} + +func TestOpenAICompatConfFromStruct_LegacyPayload(t *testing.T) { + st, err := structpb.NewStruct(map[string]any{ + "provider": "lemonade", + "endpoint": "http://legacy:8000", + "headers": map[string]any{"Authorization": "Bearer legacy"}, + "capacity": float64(4), + "max_queue": float64(10), + "queue_timeout_ms": float64(2000), + "request_timeout_ms": float64(45000), + }) + if err != nil { + t.Fatalf("structpb: %v", err) + } + cfg := openAICompatConfFromStruct(st) + if !cfg.Enabled { + t.Fatal("expected enabled=true") + } + if cfg.Provider != "lemonade" || cfg.Endpoint != "http://legacy:8000" { + t.Fatalf("config: got %+v", cfg) + } + if cfg.Headers["Authorization"] != "Bearer legacy" { + t.Fatalf("headers: got %+v", cfg.Headers) + } + if cfg.Capacity != 4 || cfg.MaxQueue != 10 || cfg.QueueTimeoutMS != 2000 || cfg.RequestTimeoutMS != 45000 { + t.Fatalf("queue config: got %+v", cfg) + } +} + func TestProviderConfsFromProto_QueueConfig(t *testing.T) { ollamaCfg := ollamaConfFromProto(&iop.OllamaAdapterConfig{ BaseUrl: "http://typed:11434", diff --git a/apps/node/internal/adapters/openai_compat/openai_compat.go b/apps/node/internal/adapters/openai_compat/openai_compat.go new file mode 100644 index 0000000..99889ae --- /dev/null +++ b/apps/node/internal/adapters/openai_compat/openai_compat.go @@ -0,0 +1,438 @@ +// Package openai_compat provides an Adapter for OpenAI-compatible inference +// servers such as Lemonade Server via the /v1/models and /v1/chat/completions +// endpoints. Unlike the vllm adapter it carries an explicit provider label, +// per-request auth/header injection and top-level option passthrough so a +// provider boundary and its config contract stay distinct. +package openai_compat + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "go.uber.org/zap" + + "iop/apps/node/internal/runtime" + "iop/packages/go/config" +) + +const Name = "openai_compat" + +type Adapter struct { + instanceName string + provider string + endpoint string + headers map[string]string + capacity int + maxQueue int + queueTimeoutMS int + requestTimeoutMS int + client *http.Client + logger *zap.Logger +} + +// New creates an OpenAI-compatible adapter. The optional instanceName is the +// registry instance key used to disambiguate multiple instances on a node. +func New(cfg config.OpenAICompatConf, logger *zap.Logger, instanceName ...string) *Adapter { + endpoint := strings.TrimRight(cfg.Endpoint, "/") + name := Name + if len(instanceName) > 0 && instanceName[0] != "" { + name = instanceName[0] + } + var headers map[string]string + if len(cfg.Headers) > 0 { + headers = make(map[string]string, len(cfg.Headers)) + for k, v := range cfg.Headers { + headers[k] = v + } + } + return &Adapter{ + instanceName: name, + provider: cfg.Provider, + endpoint: endpoint, + headers: headers, + capacity: cfg.Capacity, + maxQueue: cfg.MaxQueue, + queueTimeoutMS: cfg.QueueTimeoutMS, + requestTimeoutMS: cfg.RequestTimeoutMS, + client: &http.Client{}, + logger: logger, + } +} + +func (a *Adapter) Name() string { return Name } + +func (a *Adapter) Capabilities(ctx context.Context) (runtime.Capabilities, error) { + probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + targets, err := a.fetchTargets(probeCtx) + status := runtime.ProviderStatusAvailable + if err != nil { + status = runtime.ProviderStatusUnavailable + } + return runtime.Capabilities{ + AdapterName: Name, + InstanceKey: a.instanceName, + Targets: targets, + MaxConcurrency: effectiveCapacity(a.capacity, 8), + MaxQueue: a.maxQueue, + QueueTimeoutMS: a.queueTimeoutMS, + RequestTimeoutMS: a.requestTimeoutMS, + ProviderStatus: runtime.NormalizeProviderStatus(status), + }, nil +} + +func effectiveCapacity(capacity, defaultVal int) int { + if capacity <= 0 { + return defaultVal + } + return capacity +} + +func (a *Adapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { + if a.endpoint == "" { + return fmt.Errorf("openai_compat adapter: endpoint is required") + } + model := strings.TrimSpace(spec.Target) + if model == "" { + model = stringInput(spec.Input, "model") + } + if model == "" { + return fmt.Errorf("openai_compat adapter: target/model is required") + } + messages := messagesFromInput(spec.Input) + if len(messages) == 0 { + return fmt.Errorf("openai_compat adapter: messages are required") + } + + if err := sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeStart, + Timestamp: time.Now(), + }); err != nil { + return err + } + + reqBody := buildRequestBody(model, messages, spec.Input) + body, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("openai_compat adapter: marshal request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, joinOpenAIPath(a.endpoint, "/v1/chat/completions"), bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("openai_compat adapter: build request: %w", err) + } + a.applyHeaders(req, true) + + a.logger.Info("openai_compat adapter executing", + zap.String("run_id", spec.RunID), + zap.String("provider", a.provider), + zap.String("target", model), + zap.String("endpoint", a.endpoint), + ) + resp, err := a.client.Do(req) + if err != nil { + _ = emitError(ctx, sink, spec.RunID, fmt.Sprintf("openai_compat request failed: %v", err)) + return fmt.Errorf("openai_compat adapter: request: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + msg := readLimited(resp.Body, 4096) + if msg == "" { + msg = resp.Status + } + _ = emitError(ctx, sink, spec.RunID, fmt.Sprintf("openai_compat returned %s: %s", resp.Status, msg)) + return fmt.Errorf("openai_compat adapter: non-2xx response: %s", resp.Status) + } + + scanner := bufio.NewScanner(resp.Body) + outputTokens := 0 + finishReason := "" + var usage *runtime.UsageStats + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "data: ") { + continue + } + payload := strings.TrimPrefix(line, "data: ") + if payload == "[DONE]" { + return sink.Emit(ctx, completeEvent(spec.RunID, finishReason, usage, outputTokens)) + } + var chunk chatChunk + if err := json.Unmarshal([]byte(payload), &chunk); err != nil { + continue + } + if chunk.Usage != nil { + usage = &runtime.UsageStats{ + InputTokens: chunk.Usage.PromptTokens, + OutputTokens: chunk.Usage.CompletionTokens, + } + } + if len(chunk.Choices) == 0 { + continue + } + choice := chunk.Choices[0] + if choice.FinishReason != nil && *choice.FinishReason != "" { + finishReason = *choice.FinishReason + } + if reasoning := choice.Delta.ReasoningContent; reasoning != "" { + if err := sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeReasoningDelta, + Delta: reasoning, + Timestamp: time.Now(), + }); err != nil { + return err + } + } + if content := choice.Delta.Content; content != "" { + outputTokens += len(strings.Fields(content)) + if err := sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: spec.RunID, + Type: runtime.EventTypeDelta, + Delta: content, + Timestamp: time.Now(), + }); err != nil { + return err + } + } + } + if err := scanner.Err(); err != nil { + _ = emitError(ctx, sink, spec.RunID, fmt.Sprintf("openai_compat stream scan failed: %v", err)) + return fmt.Errorf("openai_compat adapter: scan stream: %w", err) + } + _ = emitError(ctx, sink, spec.RunID, "openai_compat stream ended without [DONE]") + return fmt.Errorf("openai_compat adapter: stream ended without [DONE]") +} + +// ProbeProvider checks endpoint availability and target presence using the +// OpenAI-compatible /v1/models endpoint, returning only the public baseline +// status values. +func (a *Adapter) ProbeProvider(ctx context.Context, target string) (runtime.ProviderProbeResult, error) { + probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + + targets, err := a.fetchTargets(probeCtx) + + result := runtime.ProviderProbeResult{ + AdapterName: Name, + InstanceKey: a.instanceName, + Target: target, + } + if a.provider != "" { + result.Metadata = map[string]string{"provider": a.provider} + } + + if err != nil { + result.Status = runtime.NormalizeProviderStatus(runtime.ProviderStatusUnavailable) + result.Detail = err.Error() + return result, nil + } + + result.Targets = targets + + if target == "" { + result.Status = runtime.NormalizeProviderStatus(runtime.ProviderStatusAvailable) + return result, nil + } + + found := false + for _, t := range targets { + if t == target { + found = true + break + } + } + + if found { + result.Status = runtime.NormalizeProviderStatus(runtime.ProviderStatusAvailable) + } else { + result.Status = runtime.NormalizeProviderStatus(runtime.ProviderStatusUnavailable) + result.Detail = fmt.Sprintf("target model %q not found in provider models", target) + } + + return result, nil +} + +func (a *Adapter) fetchTargets(ctx context.Context) ([]string, error) { + if a.endpoint == "" { + return nil, fmt.Errorf("openai_compat adapter: endpoint is required") + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, joinOpenAIPath(a.endpoint, "/v1/models"), nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + a.applyHeaders(req, false) + resp, err := a.client.Do(req) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("status code %d", resp.StatusCode) + } + var models modelsResponse + if err := json.NewDecoder(resp.Body).Decode(&models); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + out := make([]string, 0, len(models.Data)) + for _, model := range models.Data { + if model.ID != "" { + out = append(out, model.ID) + } + } + return out, nil +} + +// applyHeaders injects configured auth/secret headers. The adapter always owns +// Content-Type for JSON POSTs, so a user-provided Content-Type is ignored to +// keep the request body contract intact. +func (a *Adapter) applyHeaders(req *http.Request, jsonBody bool) { + for k, v := range a.headers { + if jsonBody && http.CanonicalHeaderKey(k) == "Content-Type" { + continue + } + req.Header.Set(k, v) + } + if jsonBody { + req.Header.Set("Content-Type", "application/json") + } +} + +func buildRequestBody(model string, messages []chatMessage, input map[string]any) map[string]any { + body := make(map[string]any) + // Copy caller options as top-level OpenAI-compatible request fields first so + // adapter-owned fields below always win over them. + if opts, ok := input["options"].(map[string]any); ok { + for k, v := range opts { + body[k] = v + } + } + // Pass through optional OpenAI-compatible fields when present in the input. + for _, key := range []string{"tools", "format", "think", "keep_alive"} { + if v, ok := input[key]; ok { + body[key] = v + } + } + body["model"] = model + body["messages"] = messages + body["stream"] = true + return body +} + +func completeEvent(runID, finishReason string, usage *runtime.UsageStats, outputTokens int) runtime.RuntimeEvent { + if usage == nil { + usage = &runtime.UsageStats{OutputTokens: outputTokens} + } + var metadata map[string]string + if finishReason != "" { + metadata = map[string]string{"finish_reason": finishReason} + } + return runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeComplete, + Message: "openai_compat chat complete", + Usage: usage, + Metadata: metadata, + Timestamp: time.Now(), + } +} + +func messagesFromInput(input map[string]any) []chatMessage { + if raw, ok := input["messages"].([]any); ok { + out := make([]chatMessage, 0, len(raw)) + for _, item := range raw { + m, ok := item.(map[string]any) + if !ok { + continue + } + role, _ := m["role"].(string) + content, _ := m["content"].(string) + role = strings.TrimSpace(role) + content = strings.TrimSpace(content) + if role == "" || content == "" { + continue + } + out = append(out, chatMessage{Role: role, Content: content}) + } + if len(out) > 0 { + return out + } + } + if prompt := strings.TrimSpace(stringInput(input, "prompt")); prompt != "" { + return []chatMessage{{Role: "user", Content: prompt}} + } + return nil +} + +func emitError(ctx context.Context, sink runtime.EventSink, runID, msg string) error { + return sink.Emit(ctx, runtime.RuntimeEvent{ + RunID: runID, + Type: runtime.EventTypeError, + Error: msg, + Timestamp: time.Now(), + }) +} + +func stringInput(input map[string]any, key string) string { + if input == nil { + return "" + } + if v, ok := input[key].(string); ok { + return v + } + return "" +} + +// joinOpenAIPath appends an OpenAI-compatible path to the endpoint. When the +// endpoint already targets the /v1 root it strips the leading /v1 from the path +// so a configured ".../v1" endpoint does not produce ".../v1/v1/...". +func joinOpenAIPath(baseURL, path string) string { + u, err := url.Parse(baseURL) + if err != nil { + return strings.TrimRight(baseURL, "/") + path + } + basePath := strings.TrimRight(u.Path, "/") + if strings.HasSuffix(basePath, "/v1") && strings.HasPrefix(path, "/v1/") { + path = strings.TrimPrefix(path, "/v1") + } + u.Path = basePath + path + return u.String() +} + +func readLimited(r io.Reader, limit int64) string { + b, _ := io.ReadAll(io.LimitReader(r, limit)) + return strings.TrimSpace(string(b)) +} + +type chatMessage struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type chatChunk struct { + Choices []struct { + Delta struct { + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + } `json:"delta"` + FinishReason *string `json:"finish_reason"` + } `json:"choices"` + Usage *struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + } `json:"usage"` +} + +type modelsResponse struct { + Data []struct { + ID string `json:"id"` + } `json:"data"` +} diff --git a/apps/node/internal/adapters/openai_compat/openai_compat_test.go b/apps/node/internal/adapters/openai_compat/openai_compat_test.go new file mode 100644 index 0000000..d1ecb01 --- /dev/null +++ b/apps/node/internal/adapters/openai_compat/openai_compat_test.go @@ -0,0 +1,330 @@ +package openai_compat + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "go.uber.org/zap" + + "iop/apps/node/internal/runtime" + "iop/packages/go/config" +) + +type fakeSink struct { + mu sync.Mutex + events []runtime.RuntimeEvent +} + +func (s *fakeSink) Emit(_ context.Context, event runtime.RuntimeEvent) error { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, event) + return nil +} + +func (s *fakeSink) all() []runtime.RuntimeEvent { + s.mu.Lock() + defer s.mu.Unlock() + return append([]runtime.RuntimeEvent(nil), s.events...) +} + +func TestOpenAICompatCapabilitiesQueryModels(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/models" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"model-a"},{"id":"model-b"}]}`)) + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{ + Provider: "lemonade", + Endpoint: server.URL, + Capacity: 10, + MaxQueue: 20, + QueueTimeoutMS: 5000, + RequestTimeoutMS: 10000, + }, zap.NewNop()) + caps, err := adapter.Capabilities(context.Background()) + if err != nil { + t.Fatalf("Capabilities failed: %v", err) + } + if caps.AdapterName != Name { + t.Fatalf("AdapterName: got %q want %q", caps.AdapterName, Name) + } + if got := strings.Join(caps.Targets, ","); got != "model-a,model-b" { + t.Fatalf("targets: got %q", got) + } + if caps.MaxConcurrency != 10 { + t.Fatalf("expected MaxConcurrency 10, got %d", caps.MaxConcurrency) + } + if caps.MaxQueue != 20 || caps.QueueTimeoutMS != 5000 || caps.RequestTimeoutMS != 10000 { + t.Fatalf("queue config: got %+v", caps) + } + if caps.ProviderStatus != runtime.ProviderStatusAvailable { + t.Fatalf("expected available, got %s", caps.ProviderStatus) + } + + adapterDefault := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + capsDefault, err := adapterDefault.Capabilities(context.Background()) + if err != nil { + t.Fatalf("Capabilities failed: %v", err) + } + if capsDefault.MaxConcurrency != 8 { + t.Fatalf("expected default MaxConcurrency 8, got %d", capsDefault.MaxConcurrency) + } +} + +func TestOpenAICompatProbeProviderAvailability(t *testing.T) { + t.Run("200_ok_and_target_hit", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/models" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"model-a"},{"id":"model-b"}]}`)) + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Provider: "lemonade", Endpoint: server.URL}, zap.NewNop()) + res, err := adapter.ProbeProvider(context.Background(), "model-a") + if err != nil { + t.Fatalf("ProbeProvider failed: %v", err) + } + if res.Status != runtime.ProviderStatusAvailable { + t.Errorf("expected available, got %s", res.Status) + } + if len(res.Targets) != 2 || res.Targets[0] != "model-a" { + t.Errorf("unexpected Targets: %+v", res.Targets) + } + if res.Metadata["provider"] != "lemonade" { + t.Errorf("expected provider metadata, got %+v", res.Metadata) + } + }) + + t.Run("200_ok_and_target_miss", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"model-a"}]}`)) + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + res, err := adapter.ProbeProvider(context.Background(), "model-b") + if err != nil { + t.Fatalf("ProbeProvider failed: %v", err) + } + if res.Status != runtime.ProviderStatusUnavailable { + t.Errorf("expected unavailable, got %s", res.Status) + } + if !strings.Contains(res.Detail, "not found") { + t.Errorf("expected 'not found' in detail, got %s", res.Detail) + } + }) + + t.Run("500_internal_error", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + res, err := adapter.ProbeProvider(context.Background(), "model-a") + if err != nil { + t.Fatalf("ProbeProvider failed: %v", err) + } + if res.Status != runtime.ProviderStatusUnavailable { + t.Errorf("expected unavailable, got %s", res.Status) + } + }) + + t.Run("empty_target", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"object":"list","data":[{"id":"model-a"}]}`)) + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + res, err := adapter.ProbeProvider(context.Background(), "") + if err != nil { + t.Fatalf("ProbeProvider failed: %v", err) + } + if res.Status != runtime.ProviderStatusAvailable { + t.Errorf("expected available, got %s", res.Status) + } + }) +} + +func TestOpenAICompatExecuteStreamsDeltasAndFinishReason(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/chat/completions" { + t.Fatalf("unexpected path %s", r.URL.Path) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"reasoning_content":"think "}}]}`) + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"hello "}}]}`) + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"world"},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":2}}`) + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + sink := &fakeSink{} + err := adapter.Execute(context.Background(), runtime.ExecutionSpec{ + RunID: "run-1", + Target: "lemonade-model", + Input: map[string]any{"prompt": "say hello"}, + }, sink) + if err != nil { + t.Fatalf("Execute failed: %v", err) + } + + events := sink.all() + // start + reasoning_delta + delta + delta + complete + if len(events) != 5 { + t.Fatalf("expected 5 events, got %d: %+v", len(events), events) + } + if events[0].Type != runtime.EventTypeStart { + t.Fatalf("expected start, got %+v", events[0]) + } + if events[1].Type != runtime.EventTypeReasoningDelta || events[1].Delta != "think " { + t.Fatalf("expected reasoning delta, got %+v", events[1]) + } + if events[2].Delta+events[3].Delta != "hello world" { + t.Fatalf("unexpected deltas: %q + %q", events[2].Delta, events[3].Delta) + } + complete := events[4] + if complete.Type != runtime.EventTypeComplete { + t.Fatalf("expected complete, got %+v", complete) + } + if complete.Metadata["finish_reason"] != "stop" { + t.Fatalf("expected finish_reason stop, got %+v", complete.Metadata) + } + if complete.Usage == nil || complete.Usage.OutputTokens != 2 || complete.Usage.InputTokens != 5 { + t.Fatalf("expected usage from chunk, got %+v", complete.Usage) + } +} + +func TestOpenAICompatExecuteSendsHeaders(t *testing.T) { + var gotAuth, gotContentType string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + gotContentType = r.Header.Get("Content-Type") + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"ok"}}]}`) + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{ + Endpoint: server.URL, + Headers: map[string]string{ + "Authorization": "Bearer secret-token", + // A user-provided Content-Type must not override the adapter-owned one. + "Content-Type": "text/plain", + }, + }, zap.NewNop()) + sink := &fakeSink{} + if err := adapter.Execute(context.Background(), runtime.ExecutionSpec{ + RunID: "run-h", + Target: "lemonade-model", + Input: map[string]any{"prompt": "hi"}, + }, sink); err != nil { + t.Fatalf("Execute failed: %v", err) + } + if gotAuth != "Bearer secret-token" { + t.Fatalf("expected Authorization header, got %q", gotAuth) + } + if gotContentType != "application/json" { + t.Fatalf("expected adapter-owned content type, got %q", gotContentType) + } +} + +func TestOpenAICompatExecutePassesOptionsAsTopLevelFields(t *testing.T) { + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("decode: %v", err) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprintf(w, "data: %s\n\n", `{"choices":[{"delta":{"content":"ok"}}]}`) + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + })) + defer server.Close() + + adapter := New(config.OpenAICompatConf{Endpoint: server.URL}, zap.NewNop()) + sink := &fakeSink{} + if err := adapter.Execute(context.Background(), runtime.ExecutionSpec{ + RunID: "run-o", + Target: "lemonade-model", + Input: map[string]any{ + "prompt": "hi", + "options": map[string]any{ + "temperature": 0.2, + "max_tokens": float64(8), + // adapter-owned fields must not be overridden by options + "model": "evil", + "stream": false, + }, + }, + }, sink); err != nil { + t.Fatalf("Execute failed: %v", err) + } + if body["temperature"] != 0.2 { + t.Fatalf("expected temperature top-level, got %v", body["temperature"]) + } + if body["max_tokens"] != float64(8) { + t.Fatalf("expected max_tokens top-level, got %v", body["max_tokens"]) + } + if body["model"] != "lemonade-model" { + t.Fatalf("options must not override model, got %v", body["model"]) + } + if body["stream"] != true { + t.Fatalf("options must not override stream, got %v", body["stream"]) + } +} + +func TestOpenAICompatExecuteRejectsEmptyEndpointOrModel(t *testing.T) { + t.Run("empty_endpoint", func(t *testing.T) { + adapter := New(config.OpenAICompatConf{}, zap.NewNop()) + err := adapter.Execute(context.Background(), runtime.ExecutionSpec{ + RunID: "r", Target: "m", Input: map[string]any{"prompt": "hi"}, + }, &fakeSink{}) + if err == nil || !strings.Contains(err.Error(), "endpoint is required") { + t.Fatalf("expected endpoint error, got %v", err) + } + }) + + t.Run("empty_model", func(t *testing.T) { + adapter := New(config.OpenAICompatConf{Endpoint: "http://localhost:8000"}, zap.NewNop()) + err := adapter.Execute(context.Background(), runtime.ExecutionSpec{ + RunID: "r", Input: map[string]any{"prompt": "hi"}, + }, &fakeSink{}) + if err == nil || !strings.Contains(err.Error(), "target/model is required") { + t.Fatalf("expected model error, got %v", err) + } + }) +} + +func TestOpenAICompatJoinOpenAIPathNoDuplicateV1(t *testing.T) { + cases := []struct { + endpoint string + path string + wantTail string + }{ + {"http://host:8000", "/v1/models", "/v1/models"}, + {"http://host:8000/", "/v1/chat/completions", "/v1/chat/completions"}, + {"http://host:8000/v1", "/v1/models", "/v1/models"}, + {"http://host:8000/v1/", "/v1/chat/completions", "/v1/chat/completions"}, + } + for _, c := range cases { + got := joinOpenAIPath(strings.TrimRight(c.endpoint, "/"), c.path) + if !strings.HasSuffix(got, c.wantTail) || strings.Contains(got, "/v1/v1") { + t.Errorf("joinOpenAIPath(%q,%q)=%q want suffix %q without /v1/v1", c.endpoint, c.path, got, c.wantTail) + } + } +}