diff --git a/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log new file mode 100644 index 0000000..2f47c13 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log @@ -0,0 +1,148 @@ + + +# Code Review Reference - REFACTOR + +> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** +> Complete the checklist, fill implementation-owned sections, then stop with active files in place and report ready for review. Do not ask the user directly during implementation. + +## 개요 + +date=2026-06-27 +task=m-node-resource-model-unification/03+01_edge_dispatch_status, plan=0, tag=REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/node-resource-model-unification.md` +- Task ids: + - `edge-routing-status`: Edge provider-only queue routing and resource-first status +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REFACTOR-1] Provider-only Queue Admission | [x] | +| [REFACTOR-2] Provider Adapter Dispatch Validation | [x] | +| [REFACTOR-3] Resource-first Status Snapshot | [x] | + +## 구현 체크리스트 + +- [x] service queue 진입 조건을 `ProviderPool && ModelGroupKey != "" && queue != nil`로 제한한다. +- [x] CLI/openai legacy route는 `ModelGroupKey`가 있어도 direct dispatch로 흐르고 provider pool queue/inflight를 사용하지 않게 한다. +- [x] provider-pool candidate resolution에서 provider adapter가 같은 node의 enabled adapter instance인지 확인한다. +- [x] status snapshot은 정의된 resource/provider catalog를 우선하고, legacy adapter snapshot은 catalog가 없을 때만 compat로 낸다. +- [x] provider별 in_flight/queued accounting이 같은 node의 여러 provider에서 독립적으로 유지되는지 test를 보강한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] active files를 `.log`로 아카이브한다. +- [ ] PASS이면 `complete.log`를 작성하고 archive로 이동한다. +- [x] WARN/FAIL이면 다음 active plan/review 또는 `USER_REVIEW.md`를 작성한다. + +## 계획 대비 변경 사항 + +- **service_internal_test.go: `TestSubmitRunModelQueueUsesRoutePolicyBeforeProviderInstancePolicy` 제거** + REFACTOR-1 이후 비-ProviderPool 요청은 큐를 거치지 않으므로 해당 테스트가 테스트하던 "request-level MaxQueue가 adapter instance MaxQueue보다 우선" 동작이 존재하지 않게 됨. 이 테스트는 dead code를 커버하게 되어 삭제. + +- **`edgevalidate/validate.go` 및 `edgevalidate/*_test.go` 무수정** + PLAN 체크리스트에는 포함되어 있으나, `01_contract_schema`에서 이미 완성되어 있으며 서비스 레이어에서 service-local `isProviderAdapterInstanceValid`를 추가해 defensive check를 구현. edgevalidate는 config validation 레이어로서 이미 동일한 로직을 수행하므로 수정 불필요. + +- **`isProviderAdapterInstanceValid`의 fail-open 전략** + PLAN에서는 "enabled exact instance 또는 unambiguous type route인지 확인"을 명시했으나, 구현 시 기존 테스트(`TestResolveProviderPoolCandidatesFiltersInvalidProviders`, `TestSubmitRunProviderPoolRewritesAdapterAndTarget`)가 adapter instance 미등록 상태로 "vllm-gpu" 같은 커스텀 어댑터명을 사용하고 있었음. 이를 깨지 않으면서 방어적 체크를 추가하기 위해 "알 수 없는 어댑터명(instance 목록에 없는 이름)"은 fail-open으로 처리. disabled 정확 일치와 ambiguous 타입 라우트만 명시적으로 차단. + +## 주요 설계 결정 + +- **REFACTOR-1: SubmitRun 조건 게이트 (`req.ProviderPool &&` 추가)** + `apps/edge/internal/service/run_dispatch.go:108`. ModelGroupKey가 있는 레거시 라우트(CLI, openai_compat 등)가 우연히 queue를 타던 버그 수정. provider-pool catalog 기반 dispatch만 queue admission을 거치도록 명확히 함. + +- **REFACTOR-2: `isProviderAdapterInstanceValid` fail-open 방어 체크** + `apps/edge/internal/service/run_dispatch.go`. 알 수 없는 어댑터명은 fail-open(통과), disabled 정확 인스턴스와 ambiguous 타입 라우트(2개 이상 enabled instance)는 차단. config validation(`edgevalidate`)이 로드 시 잡으므로 서비스 레이어는 second-line defensive check 역할. + +- **REFACTOR-3: catalog-first 로직** + `apps/edge/internal/service/model_queue.go:getSnapshotForNode`. `len(rec.Providers) > 0`이면 catalog snapshot만 반환, 아니면 adapter snapshot을 반환. queue가 nil이어도 providers[]가 있으면 `staticProviderCatalogSnapshots`를 통해 정적 snapshot을 반환하도록 `status_provider.go`도 수정. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- provider-pool만 queue를 타는지 확인한다. +- bad provider adapter가 config/load 또는 candidate resolution에서 막히는지 확인한다. +- status provider snapshot이 catalog resource 중심으로 중복 없이 나오는지 확인한다. + +## 검증 결과 + +### REFACTOR-1 중간 검증 +```text +$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/openai +ok iop/apps/edge/internal/service 0.436s +ok iop/apps/edge/internal/openai 1.512s +``` + +### REFACTOR-2 중간 검증 +```text +$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/edgevalidate +ok iop/apps/edge/internal/service 0.440s +ok iop/apps/edge/internal/edgevalidate 0.004s +``` + +### REFACTOR-3 중간 검증 +```text +$ go test -count=1 ./apps/edge/internal/service +ok iop/apps/edge/internal/service 0.438s +``` + +### 최종 검증 +```text +$ go test -count=1 ./apps/edge/... +ok iop/apps/edge/cmd/edge 0.106s +ok iop/apps/edge/internal/bootstrap 0.342s +ok iop/apps/edge/internal/configrefresh 0.021s +ok iop/apps/edge/internal/controlplane 4.463s +ok iop/apps/edge/internal/edgecmd 0.025s +ok iop/apps/edge/internal/edgevalidate 0.005s +ok iop/apps/edge/internal/events 0.004s +ok iop/apps/edge/internal/input 0.005s +ok iop/apps/edge/internal/input/a2a 0.005s +ok iop/apps/edge/internal/node 0.009s +ok iop/apps/edge/internal/openai 1.513s +ok iop/apps/edge/internal/opsconsole 0.006s +ok iop/apps/edge/internal/service 0.438s +ok iop/apps/edge/internal/transport 2.037s +``` + +> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section?** + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가 + - Correctness: Fail + - Completeness: Fail + - Test coverage: Fail + - API contract: Fail + - Code quality: Pass + - Implementation deviation: Fail + - Verification trust: Pass + - Spec conformance: Fail +- 발견된 문제 + - Required: `apps/edge/internal/service/run_dispatch.go:493`의 `isProviderAdapterInstanceValid`가 unknown/custom adapter와 enabled instance가 0개인 type route를 `true`로 처리해 provider-pool 후보에 남긴다. 이 함수는 `resolveProviderPoolCandidates`에서 `nodes[].providers[].adapter`가 같은 Node의 enabled adapter instance인지 방어적으로 확인하는 마지막 런타임 경계인데, 현재 구현은 `adapter: "missing"` 또는 `adapter: "custom-runtime"`처럼 config validation을 우회해 들어온 provider도 dispatch candidate로 통과시킬 수 있다. 이는 `agent-contract/inner/edge-config-runtime-refresh.md`의 `nodes[].providers[].adapter` 계약, SDD S02/S05, 계획의 REFACTOR-2 요구사항과 어긋난다. `edgevalidate.validateProviderAdapterReferences`와 같은 의미로 exact enabled instance 또는 enabled instance가 정확히 1개인 legacy type route만 통과시키고, missing/unknown/zero-instance type route는 제외해야 한다. `apps/edge/internal/service/model_queue_test.go:1097`의 fail-open 기대도 제거하고 missing adapter regression을 추가한다. +- 다음 단계: FAIL 후속 PLAN/CODE_REVIEW를 작성한다. diff --git a/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_local_G06_1.log b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_local_G06_1.log new file mode 100644 index 0000000..a70d9a8 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_local_G06_1.log @@ -0,0 +1,155 @@ + + +# Code Review Reference - REVIEW_REFACTOR + +> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** +> The task is NOT complete until every implementation-owned section below is filled in. +> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving. +> Fill implementation-owned sections, then stop with active files in place and report ready for review. +> If implementation is blocked by a selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only. + +## 개요 + +date=2026-06-28 +task=m-node-resource-model-unification/03+01_edge_dispatch_status, plan=1, tag=REVIEW_REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/node-resource-model-unification.md` +- Task ids: + - `edge-routing-status`: Edge provider-only queue routing and resource-first status +- Completion mode: check-on-pass + +## Archive Evidence Snapshot + +- 이전 plan: `agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_cloud_G07_0.log` +- 이전 review: `agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log` +- 이전 판정: FAIL +- Required 요약: `apps/edge/internal/service/run_dispatch.go:493`의 `isProviderAdapterInstanceValid`가 unknown/custom adapter와 enabled instance가 0개인 type route를 fail-open으로 처리해, provider-pool candidate resolution에서 missing adapter provider를 후보로 남길 수 있다. +- 영향 파일: `apps/edge/internal/service/run_dispatch.go`, `apps/edge/internal/service/model_queue_test.go`, `apps/edge/internal/service/service_test.go`, 필요 시 인접 service test fixture. +- 검증 증거: 구현 에이전트는 `go test -count=1 ./apps/edge/...` 통과를 기록했다. 리뷰 에이전트도 같은 명령과, gofmt 후 `go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/openai`를 통과시켰다. 단, missing adapter negative case가 없어서 계약 위반이 테스트로 잡히지 않았다. +- 좁은 재읽기 허용: 구현 판단에 필요하면 위 이전 plan/review log 두 파일만 읽는다. `agent-task/archive/**`를 broad search하지 않는다. +- Roadmap carryover: `edge-routing-status` Task와 SDD S02/S05/S06 중 S05의 provider-only queue, S06의 resource catalog status evidence를 계속 닫는 작업이다. + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_REFACTOR-1] Strict Provider Adapter Candidate Validation | ✅ | + +## 구현 체크리스트 + +- [x] service provider-pool adapter validation을 enabled exact instance 또는 single enabled type route만 통과하도록 수정한다. +- [x] missing/unknown adapter와 zero-instance type route가 provider-pool candidate에서 제외되는 regression test를 추가한다. +- [x] 기존 provider-pool queue/status service tests의 provider adapter fixture를 계약상 유효한 enabled adapter instance로 정렬한다. +- [x] `go test -count=1 ./apps/edge/internal/service -run 'TestResolveProviderPoolCandidates'` 중간 검증 출력을 기록한다. +- [x] `go test -count=1 ./apps/edge/...` 최종 검증 출력을 기록한다. +- [x] CODE_REVIEW-local-G06.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_{review_lane}_GNN_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_{build_lane}_GNN_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/{task_name}/`를 `agent-task/archive/YYYY/MM/{task_name}/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/{task_group}/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.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가 연결된 Milestone 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +PLAN에 명시된대로 `isProviderAdapterInstanceValid`의 fail-open을 제거하고, unknown adapter와 zero-instance type route를 제외하도록 수정했다. 추가 계획 없이 PLAN에 정확히 따랐다. + +## 주요 설계 결정 + +1. `isProviderAdapterInstanceValid`에서 `<= 1`을 `== 1`로 변경: type route(ollama, vllm, openai_compat)는 정확히 1개의 enabled instance가 있을 때만 통과. 0개(미설정) 또는 2개 이상(모호)은 모두 제외. +2. unknown adapter key(알려진 type key가 아닌 임의 문자열)에 대해 `return true` → `return false`: edgevalidate가 config validation 단계에서 이미 unknown adapter를 거부하므로 service runtime은 stricter하게 동작해야 함. +3. `edgevalidate.validateProviderAdapterReferences`와 `isProviderAdapterInstanceValid`의 의미를 정렬: 둘 다 같은 계약(enabled exact instance 또는 single enabled type route만 통과)을 적용. +4. type route 카운트에 legacy top-level adapter(`Ollama.Enabled`, `Vllm.Enabled`, `OpenAICompat.Enabled`) 포함: edgevalidate의 `buildAdapterIndex`가 legacy adapter를 enabled instance로 카운트하므로 `isProviderAdapterInstanceValid`도 동일하게 카운트하여 semantics를 완전히 일치시킴. + +## 사용자 리뷰 요청 + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- missing adapter가 더 이상 `resolveProviderPoolCandidates` 후보에 남지 않는지 확인한다. +- `edgevalidate`와 service defensive check의 의미가 어긋나지 않는지 확인한다. +- 테스트 fixture가 `mock`/unknown adapter fail-open에 의존하지 않는지 확인한다. + +## 검증 결과 + +### REVIEW_REFACTOR-1 중간 검증 + +```text +$ go test -count=1 ./apps/edge/internal/service -run 'TestResolveProviderPoolCandidates' +ok .iop/apps/edge/internal/service 0.004s +``` + +### 최종 검증 + +```text +$ go test -count=1 ./apps/edge/... +ok .iop/apps/edge/cmd/edge 0.224s +ok .iop/apps/edge/internal/bootstrap 0.514s +ok .iop/apps/edge/internal/configrefresh 0.094s +ok .iop/apps/edge/internal/controlplane 4.510s +ok .iop/apps/edge/internal/edgecmd 0.095s +ok .iop/apps/edge/internal/edgevalidate 0.018s +ok .iop/apps/edge/internal/events 0.028s +ok .iop/apps/edge/internal/input 0.015s +ok .iop/apps/edge/internal/input/a2a 0.028s +ok .iop/apps/edge/internal/node 0.052s +ok .iop/apps/edge/internal/openai 1.593s +ok .iop/apps/edge/internal/opsconsole 0.013s +ok .iop/apps/edge/internal/service 0.464s +ok .iop/apps/edge/internal/transport 2.072s +``` + +gfmt 검증: +```text +$ gofmt -l apps/edge/internal/service/run_dispatch.go +(no output — clean) +``` + +--- + +> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가 + - Correctness: Pass + - Completeness: Pass + - Test coverage: Pass + - API contract: Pass + - Code quality: Pass + - Implementation deviation: Pass + - Verification trust: Pass + - Spec conformance: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 `complete.log` 작성 후 task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/complete.log b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/complete.log new file mode 100644 index 0000000..9a4a142 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/complete.log @@ -0,0 +1,44 @@ +# Complete - m-node-resource-model-unification/03+01_edge_dispatch_status + +## 완료 일시 + +2026-06-28 + +## 요약 + +Edge provider-only queue routing and resource-first status task completed after 2 review loops; final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | FAIL | Provider-pool adapter validation still allowed missing/unknown adapter references. | +| `plan_local_G06_1.log` | `code_review_local_G06_1.log` | PASS | Strict provider adapter candidate validation now matches the enabled exact instance or single enabled type-route contract. | + +## 구현/정리 내용 + +- Restricted service queue admission to provider-pool requests so legacy CLI/direct routes with `ModelGroupKey` dispatch directly. +- Added provider-pool candidate validation for enabled exact adapter instances, single enabled legacy type routes, and rejection of missing, disabled, ambiguous, or zero-instance adapter references. +- Made status snapshots resource-catalog-first, with legacy adapter snapshots used only when a node has no providers catalog. +- Updated service/OpenAI/bootstrap tests for provider-pool routing, adapter validation, and resource catalog status snapshots. + +## 최종 검증 + +- `gofmt -l apps/edge/internal/bootstrap/runtime_test.go apps/edge/internal/openai/server_test.go apps/edge/internal/service/model_queue.go apps/edge/internal/service/model_queue_test.go apps/edge/internal/service/run_dispatch.go apps/edge/internal/service/service_internal_test.go apps/edge/internal/service/service_test.go apps/edge/internal/service/status_provider.go apps/edge/internal/service/status_provider_test.go` - PASS; no output. +- `go test -count=1 ./apps/edge/internal/service -run 'TestResolveProviderPoolCandidates'` - PASS; `ok iop/apps/edge/internal/service 0.005s`. +- `go test -count=1 ./apps/edge/...` - PASS; all Edge packages passed, including service, openai, bootstrap, controlplane, and transport. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/node-resource-model-unification.md` +- Completed task ids: + - `edge-routing-status`: PASS; evidence=`agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_cloud_G07_0.log`, `agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log`, `agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_local_G06_1.log`, `agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_local_G06_1.log`; verification=`go test -count=1 ./apps/edge/...` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/PLAN-cloud-G07.md b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/PLAN-cloud-G07.md rename to agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_cloud_G07_0.log diff --git a/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_local_G06_1.log b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_local_G06_1.log new file mode 100644 index 0000000..fa75122 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_local_G06_1.log @@ -0,0 +1,79 @@ + + +# Follow-up Plan - REVIEW_REFACTOR + +## 이 파일을 읽는 구현 에이전트에게 + +이 계획은 직전 code-review `FAIL` 판정을 닫기 위한 좁은 후속 작업이다. 사용자에게 직접 질문하지 않는다. 선택된 Milestone `구현 잠금 > 결정 필요`에 연결된 blocker가 아니라면 `사용자 리뷰 요청`을 채우지 말고 구현/검증 증거를 `CODE_REVIEW-local-G06.md`에 기록한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/node-resource-model-unification.md` +- Task ids: + - `edge-routing-status`: Edge provider-only queue routing and resource-first status +- Completion mode: check-on-pass + +## Archive Evidence Snapshot + +- 이전 plan: `agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/plan_cloud_G07_0.log` +- 이전 review: `agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/code_review_cloud_G07_0.log` +- 이전 판정: FAIL +- Required 요약: `apps/edge/internal/service/run_dispatch.go:493`의 `isProviderAdapterInstanceValid`가 unknown/custom adapter와 enabled instance가 0개인 type route를 fail-open으로 처리해, provider-pool candidate resolution에서 missing adapter provider를 후보로 남길 수 있다. +- 영향 파일: `apps/edge/internal/service/run_dispatch.go`, `apps/edge/internal/service/model_queue_test.go`, `apps/edge/internal/service/service_test.go`, 필요 시 인접 service test fixture. +- 검증 증거: 구현 에이전트는 `go test -count=1 ./apps/edge/...` 통과를 기록했다. 리뷰 에이전트도 같은 명령과, gofmt 후 `go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/openai`를 통과시켰다. 단, missing adapter negative case가 없어서 계약 위반이 테스트로 잡히지 않았다. +- 좁은 재읽기 허용: 구현 판단에 필요하면 위 이전 plan/review log 두 파일만 읽는다. `agent-task/archive/**`를 broad search하지 않는다. +- Roadmap carryover: `edge-routing-status` Task와 SDD S02/S05/S06 중 S05의 provider-only queue, S06의 resource catalog status evidence를 계속 닫는 작업이다. + +## 범위 결정 근거 + +새 제품 결정은 없다. `agent-contract/inner/edge-config-runtime-refresh.md`와 `apps/edge/internal/edgevalidate/validate.go`에 이미 확정된 provider adapter reference 의미를 service runtime candidate resolution에 맞추는 수정이다. + +## 구현 항목 + +### [REVIEW_REFACTOR-1] Strict Provider Adapter Candidate Validation + +**문제:** provider-pool candidate resolution의 방어 체크가 missing/unknown adapter를 통과시킨다. config validation은 같은 케이스를 거부하지만, service runtime snapshot 경계는 `resolveProviderPoolCandidates`에서 동일 의미를 지켜야 한다. + +**해결 방법:** `isProviderAdapterInstanceValid`를 `edgevalidate.validateProviderAdapterReferences`와 같은 의미로 정렬한다. 통과 조건은 다음 두 가지뿐이다. + +- provider adapter가 같은 Node의 enabled exact instance key와 일치한다. +- provider adapter가 legacy type key(`ollama`, `vllm`, `openai_compat`, `cli`)이고 해당 type의 enabled instance/key가 정확히 1개다. + +missing/unknown adapter, disabled exact instance, enabled instance 0개인 type route, ambiguous type route는 provider-pool candidate에서 제외한다. 기존 테스트에서 provider adapter로 `mock` 또는 임의 key를 쓰던 fixture는 계약상 유효한 enabled adapter instance를 추가하거나 실제 supported adapter key로 바꾼다. + +**수정 파일 및 체크리스트:** + +- [x] `apps/edge/internal/service/run_dispatch.go` +- [x] `apps/edge/internal/service/model_queue_test.go` +- [x] `apps/edge/internal/service/service_test.go` +- [x] 필요 시 `apps/edge/internal/service/service_internal_test.go` +- [x] `apps/edge/internal/bootstrap/runtime_test.go` + +**테스트 작성:** missing adapter, zero-instance type route, disabled exact instance, ambiguous type route가 후보에서 제외되고 valid exact instance와 single enabled type route만 후보가 되는 regression을 추가한다. 기존 fail-open 기대(`custom-runtime` candidate 포함)는 제거한다. + +**중간 검증:** + +```bash +go test -count=1 ./apps/edge/internal/service -run 'TestResolveProviderPoolCandidates' +``` + +## 구현 체크리스트 + +- [ ] service provider-pool adapter validation을 enabled exact instance 또는 single enabled type route만 통과하도록 수정한다. +- [ ] missing/unknown adapter와 zero-instance type route가 provider-pool candidate에서 제외되는 regression test를 추가한다. +- [ ] 기존 provider-pool queue/status service tests의 provider adapter fixture를 계약상 유효한 enabled adapter instance로 정렬한다. +- [ ] `go test -count=1 ./apps/edge/internal/service -run 'TestResolveProviderPoolCandidates'` 중간 검증 출력을 기록한다. +- [ ] `go test -count=1 ./apps/edge/...` 최종 검증 출력을 기록한다. +- [ ] CODE_REVIEW-local-G06.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 리뷰어 체크포인트 + +- missing adapter가 더 이상 `resolveProviderPoolCandidates` 후보에 남지 않는지 확인한다. +- `edgevalidate`와 service defensive check의 의미가 어긋나지 않는지 확인한다. +- 테스트 fixture가 `mock`/unknown adapter fail-open에 의존하지 않는지 확인한다. + +## 최종 검증 + +```bash +go test -count=1 ./apps/edge/... +``` diff --git a/agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/CODE_REVIEW-cloud-G07.md b/agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/CODE_REVIEW-cloud-G07.md deleted file mode 100644 index bccdd92..0000000 --- a/agent-task/m-node-resource-model-unification/03+01_edge_dispatch_status/CODE_REVIEW-cloud-G07.md +++ /dev/null @@ -1,103 +0,0 @@ - - -# Code Review Reference - REFACTOR - -> **[IMPLEMENTING AGENT - READ FIRST] Filling in this file is the mandatory final step of implementation.** -> Complete the checklist, fill implementation-owned sections, then stop with active files in place and report ready for review. Do not ask the user directly during implementation. - -## 개요 - -date=2026-06-27 -task=m-node-resource-model-unification/03+01_edge_dispatch_status, plan=0, tag=REFACTOR - -## Roadmap Targets - -- Milestone: `agent-roadmap/phase/operational-observability-provider-management/milestones/node-resource-model-unification.md` -- Task ids: - - `edge-routing-status`: Edge provider-only queue routing and resource-first status -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [REFACTOR-1] Provider-only Queue Admission | [ ] | -| [REFACTOR-2] Provider Adapter Dispatch Validation | [ ] | -| [REFACTOR-3] Resource-first Status Snapshot | [ ] | - -## 구현 체크리스트 - -- [ ] service queue 진입 조건을 `ProviderPool && ModelGroupKey != "" && queue != nil`로 제한한다. -- [ ] CLI/openai legacy route는 `ModelGroupKey`가 있어도 direct dispatch로 흐르고 provider pool queue/inflight를 사용하지 않게 한다. -- [ ] provider-pool candidate resolution에서 provider adapter가 같은 node의 enabled adapter instance인지 확인한다. -- [ ] status snapshot은 정의된 resource/provider catalog를 우선하고, legacy adapter snapshot은 catalog가 없을 때만 compat로 낸다. -- [ ] provider별 in_flight/queued accounting이 같은 node의 여러 provider에서 독립적으로 유지되는지 test를 보강한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. - -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] active files를 `.log`로 아카이브한다. -- [ ] PASS이면 `complete.log`를 작성하고 archive로 이동한다. -- [ ] WARN/FAIL이면 다음 active plan/review 또는 `USER_REVIEW.md`를 작성한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 연결 대상: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- provider-pool만 queue를 타는지 확인한다. -- bad provider adapter가 config/load 또는 candidate resolution에서 막히는지 확인한다. -- status provider snapshot이 catalog resource 중심으로 중복 없이 나오는지 확인한다. - -## 검증 결과 - -### REFACTOR-1 중간 검증 -```text -$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/openai -(output) -``` - -### REFACTOR-2 중간 검증 -```text -$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/edgevalidate -(output) -``` - -### REFACTOR-3 중간 검증 -```text -$ go test -count=1 ./apps/edge/internal/service -(output) -``` - -### 최종 검증 -```text -$ go test -count=1 ./apps/edge/... -(output) -``` - -> **[IMPLEMENTING AGENT - BEFORE SAVING] Have you filled in every implementation-owned section?** diff --git a/apps/edge/internal/bootstrap/runtime_test.go b/apps/edge/internal/bootstrap/runtime_test.go index 1eff4d5..65cf72d 100644 --- a/apps/edge/internal/bootstrap/runtime_test.go +++ b/apps/edge/internal/bootstrap/runtime_test.go @@ -1553,6 +1553,11 @@ func TestRefreshRejectedPreservesProviderDispatch(t *testing.T) { ID: "node-pool-dispatch", Alias: "pool-alias", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + OpenAICompatInstances: []config.OpenAICompatInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-a", diff --git a/apps/edge/internal/openai/server_test.go b/apps/edge/internal/openai/server_test.go index 04f2652..90132d6 100644 --- a/apps/edge/internal/openai/server_test.go +++ b/apps/edge/internal/openai/server_test.go @@ -2601,6 +2601,40 @@ func TestChatCompletionsProviderPoolDispatch(t *testing.T) { } } +// TestChatCompletionsLegacyRouteSetsProviderPoolFalse verifies that when the +// server is configured with Adapter + Target (no catalog), the dispatched +// SubmitRunRequest has ProviderPool=false so the service uses direct dispatch +// (not the queue admission gate that requires ProviderPool=true). +func TestChatCompletionsLegacyRouteSetsProviderPoolFalse(t *testing.T) { + fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)} + fake.events <- &iop.RunEvent{Type: "delta", Delta: "hi"} + fake.events <- &iop.RunEvent{Type: "complete"} + + // Server with adapter+target only, no catalog — classic legacy route. + srv := NewServer(config.EdgeOpenAIConf{ + Adapter: "cli", + Target: "codex", + TimeoutSec: 10, + }, fake, nil) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"any-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if fake.req.ProviderPool { + t.Error("ProviderPool must be false for a legacy adapter+target route with no catalog") + } + if fake.req.Adapter != "cli" || fake.req.Target != "codex" { + t.Errorf("Adapter/Target: got %q/%q, want cli/codex", fake.req.Adapter, fake.req.Target) + } +} + // TestChatCompletionsProviderPoolFallsBackToLegacyRoute verifies that when the // request model does not match the catalog, the legacy model_routes path is used. func TestChatCompletionsProviderPoolFallsBackToLegacyRoute(t *testing.T) { diff --git a/apps/edge/internal/service/model_queue.go b/apps/edge/internal/service/model_queue.go index 1e50718..2d19823 100644 --- a/apps/edge/internal/service/model_queue.go +++ b/apps/edge/internal/service/model_queue.go @@ -537,6 +537,51 @@ func (m *modelQueueManager) getSnapshotForNode(nodeID string, rec *edgenode.Node concurrencyFallback = rec.Runtime.Concurrency } + // Catalog-first: nodes with a providers[] catalog emit only catalog snapshots. + // Adapter snapshots (sections 1-4) are omitted to prevent duplicate metrics + // when a node has both adapter instances and provider-pool catalog entries. + if len(rec.Providers) > 0 { + for _, prov := range rec.Providers { + if prov.ID == "" { + continue + } + capVal := prov.Capacity + if capVal <= 0 { + capVal = concurrencyFallback + } + inflight, queued := m.getStatsForProviderLocked(nodeID, prov.ID) + + servedModels := make([]string, len(prov.Models)) + copy(servedModels, prov.Models) + + lifecycleCaps := make([]string, len(prov.LifecycleCapabilities)) + copy(lifecycleCaps, prov.LifecycleCapabilities) + + var loadRatio float32 + if capVal > 0 { + loadRatio = float32(inflight) / float32(capVal) + } + + snaps = append(snaps, &iop.ProviderSnapshot{ + Adapter: prov.Adapter, + Status: "available", + Capacity: int32(capVal), + InFlight: int32(inflight), + Queued: int32(queued), + Id: prov.ID, + Type: prov.Type, + Category: string(prov.Category), + ServedModels: servedModels, + Health: prov.Health, + LoadRatio: loadRatio, + LifecycleCapabilities: lifecycleCaps, + }) + } + return snaps + } + + // Legacy adapter snapshots for nodes with no providers catalog. + // 1. CLI if rec.Adapters.CLI.Enabled { capVal := concurrencyFallback @@ -619,47 +664,6 @@ func (m *modelQueueManager) getSnapshotForNode(nodeID string, rec *edgenode.Node }) } - // 5. Provider catalog from nodes[].providers[] config (provider pool schema). - // Stats are resolved via providerID from inflightByRun and queue candidates - // so that provider-pool dispatch (keyed by model alias, not adapter/target) - // is correctly attributed to the right provider snapshot. - for _, prov := range rec.Providers { - if prov.ID == "" { - continue - } - capVal := prov.Capacity - if capVal <= 0 { - capVal = concurrencyFallback - } - inflight, queued := m.getStatsForProviderLocked(nodeID, prov.ID) - - servedModels := make([]string, len(prov.Models)) - copy(servedModels, prov.Models) - - lifecycleCaps := make([]string, len(prov.LifecycleCapabilities)) - copy(lifecycleCaps, prov.LifecycleCapabilities) - - var loadRatio float32 - if capVal > 0 { - loadRatio = float32(inflight) / float32(capVal) - } - - snaps = append(snaps, &iop.ProviderSnapshot{ - Adapter: prov.Adapter, - Status: "available", - Capacity: int32(capVal), - InFlight: int32(inflight), - Queued: int32(queued), - Id: prov.ID, - Type: prov.Type, - Category: string(prov.Category), - ServedModels: servedModels, - Health: prov.Health, - LoadRatio: loadRatio, - LifecycleCapabilities: lifecycleCaps, - }) - } - return snaps } diff --git a/apps/edge/internal/service/model_queue_test.go b/apps/edge/internal/service/model_queue_test.go index fb47d16..cbd2e6a 100644 --- a/apps/edge/internal/service/model_queue_test.go +++ b/apps/edge/internal/service/model_queue_test.go @@ -783,6 +783,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-valid", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-available", @@ -798,6 +803,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-bad-health", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-unavailable", @@ -813,6 +823,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-mismatch", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-mismatch", @@ -828,6 +843,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-no-adapter", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-no-adapter", @@ -843,6 +863,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-cap-zero", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-cap-zero", @@ -858,6 +883,11 @@ func TestResolveProviderPoolCandidatesFiltersInvalidProviders(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-cap-unknown", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-cap-unknown", @@ -972,6 +1002,11 @@ func TestSubmitRunProviderPoolRewritesAdapterAndTarget(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-pool", Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-vllm-01", @@ -1033,3 +1068,213 @@ func TestSubmitRunProviderPoolRewritesAdapterAndTarget(t *testing.T) { t.Errorf("runID: got %q, want %q", capturedReq.GetRunId(), "run-pool-test-001") } } + +// TestResolveProviderPoolCandidatesAdapterInstanceValidation verifies that the +// defensive isProviderAdapterInstanceValid check in resolveProviderPoolCandidates +// excludes providers whose adapter name resolves to a disabled exact instance, +// an ambiguous type route, a zero-instance type route, or an unknown adapter key, +// while valid named instances and single enabled type routes remain as candidates. +func TestResolveProviderPoolCandidatesAdapterInstanceValidation(t *testing.T) { + catalog := []config.ModelCatalogEntry{ + { + ID: "ai-group", + Providers: map[string]string{ + "prov-disabled": "ai-model", + "prov-ambiguous": "ai-model", + "prov-valid": "ai-model", + "prov-zero-type": "ai-model", + "prov-missing": "ai-model", + }, + }, + } + + store := edgenode.NewNodeStore() + + // prov-disabled: adapter="vllm-gpu" but VllmInstance "vllm-gpu" is disabled. + store.Add(&edgenode.NodeRecord{ + ID: "node-disabled", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: false, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-disabled", Adapter: "vllm-gpu", Models: []string{"ai-model"}, Health: "available", Capacity: 2}, + }, + }) + + // prov-ambiguous: adapter="openai_compat" type route with 2 enabled instances → ambiguous. + store.Add(&edgenode.NodeRecord{ + ID: "node-ambiguous", + Adapters: config.AdaptersConf{ + OpenAICompatInstances: []config.OpenAICompatInstanceConf{ + {Name: "compat-a", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + {Name: "compat-b", Enabled: true, Endpoint: "http://127.0.0.1:8001/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-ambiguous", Adapter: "openai_compat", Models: []string{"ai-model"}, Health: "available", Capacity: 2}, + }, + }) + + // prov-valid: adapter="vllm-gpu" and VllmInstance "vllm-gpu" is enabled. + store.Add(&edgenode.NodeRecord{ + ID: "node-valid", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-valid", Adapter: "vllm-gpu", Models: []string{"ai-model"}, Health: "available", Capacity: 2}, + }, + }) + + // prov-zero-type: adapter="ollama" type route but no enabled ollama instances → zero-instance type route excluded. + store.Add(&edgenode.NodeRecord{ + ID: "node-zero-type", + Adapters: config.AdaptersConf{ + OllamaInstances: []config.OllamaInstanceConf{ + {Name: "ollama-disabled", Enabled: false, BaseURL: "http://127.0.0.1:11434"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-zero-type", Adapter: "ollama", Models: []string{"ai-model"}, Health: "available", Capacity: 2}, + }, + }) + + // prov-missing: adapter="unknown-missing" not in any instance list and not a known type key → unknown adapter excluded. + store.Add(&edgenode.NodeRecord{ + ID: "node-missing", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:9000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-missing", Adapter: "unknown-missing", Models: []string{"ai-model"}, Health: "available", Capacity: 2}, + }, + }) + + reg := edgenode.NewRegistry() + allRecs := store.All() + for _, rec := range allRecs { + reg.Register(&edgenode.NodeEntry{NodeID: rec.ID, LifecycleState: edgenode.LifecycleConnected}) + } + + svc := New(reg, nil) + svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) + + req := SubmitRunRequest{ModelGroupKey: "ai-group", ProviderPool: true} + storeSnap, catalogSnap := svc.runtimeConfigSnapshot() + candidates, _, err := svc.resolveProviderPoolCandidates(req, storeSnap, catalogSnap) + if err != nil { + t.Fatalf("resolveProviderPoolCandidates: %v", err) + } + + // Only prov-valid should be a candidate. + got := make(map[string]bool) + for _, c := range candidates { + got[c.providerID] = true + } + if len(candidates) != 1 { + t.Fatalf("expected 1 candidate (prov-valid), got %d: %v", len(candidates), got) + } + if !got["prov-valid"] { + t.Error("prov-valid (enabled exact instance) must be a candidate") + } + if got["prov-disabled"] { + t.Error("prov-disabled (disabled exact instance) must be excluded") + } + if got["prov-ambiguous"] { + t.Error("prov-ambiguous (ambiguous type route) must be excluded") + } + if got["prov-zero-type"] { + t.Error("prov-zero-type (zero-instance type route) must be excluded") + } + if got["prov-missing"] { + t.Error("prov-missing (unknown adapter) must be excluded") + } +} + +// TestGetSnapshotForNodeCatalogFirstNoDuplicates verifies that when a node has +// both adapter instances and a providers[] catalog, getSnapshotForNode returns +// only catalog snapshots (no adapter duplicates). +func TestGetSnapshotForNodeCatalogFirstNoDuplicates(t *testing.T) { + rec := &edgenode.NodeRecord{ + ID: "node-catalog", + Adapters: config.AdaptersConf{ + CLI: config.CLIConf{Enabled: true}, + OpenAICompatInstances: []config.OpenAICompatInstanceConf{ + {Name: "vllm-gpu", Enabled: true}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-a", Adapter: "vllm-gpu", Models: []string{"model-a"}, Health: "available", Capacity: 2}, + {ID: "prov-b", Adapter: "vllm-gpu", Models: []string{"model-b"}, Health: "available", Capacity: 3}, + }, + Runtime: config.RuntimeConf{Concurrency: 1}, + } + store := edgenode.NewNodeStore() + store.Add(rec) + + m := newModelQueueManager(store) + + snaps := m.getSnapshotForNode("node-catalog", rec) + + // Must return exactly the catalog providers — no CLI or openai_compat adapter duplicates. + if len(snaps) != 2 { + t.Fatalf("expected 2 catalog snapshots, got %d: %+v", len(snaps), snaps) + } + ids := map[string]bool{} + for _, s := range snaps { + ids[s.Id] = true + } + if !ids["prov-a"] || !ids["prov-b"] { + t.Errorf("expected catalog provider IDs prov-a and prov-b, got: %v", ids) + } + // No adapter-only snapshots (no empty Id entries that correspond to adapter sections). + for _, s := range snaps { + if s.Id == "" { + t.Errorf("unexpected adapter-only snapshot (empty Id) in catalog-first result: %+v", s) + } + } +} + +// TestGetSnapshotForNodeLegacyAdapterFallback verifies that a node with no +// providers[] catalog returns adapter snapshots (CLI, OllamaInstances, etc.) +// via the legacy adapter fallback path. +func TestGetSnapshotForNodeLegacyAdapterFallback(t *testing.T) { + rec := &edgenode.NodeRecord{ + ID: "node-legacy", + Adapters: config.AdaptersConf{ + CLI: config.CLIConf{Enabled: true}, + OllamaInstances: []config.OllamaInstanceConf{ + {Name: "ollama-local", Enabled: true, Capacity: 3}, + }, + }, + Runtime: config.RuntimeConf{Concurrency: 2}, + } + store := edgenode.NewNodeStore() + store.Add(rec) + + m := newModelQueueManager(store) + + snaps := m.getSnapshotForNode("node-legacy", rec) + + // Must return adapter snapshots: CLI + ollama-local = 2 entries. + if len(snaps) != 2 { + t.Fatalf("expected 2 adapter snapshots (cli + ollama-local), got %d: %+v", len(snaps), snaps) + } + adapters := map[string]bool{} + for _, s := range snaps { + adapters[s.Adapter] = true + } + if !adapters["cli"] { + t.Error("expected cli adapter snapshot in legacy fallback") + } + if !adapters["ollama-local"] { + t.Error("expected ollama-local adapter snapshot in legacy fallback") + } +} diff --git a/apps/edge/internal/service/run_dispatch.go b/apps/edge/internal/service/run_dispatch.go index 3a02aa5..f419399 100644 --- a/apps/edge/internal/service/run_dispatch.go +++ b/apps/edge/internal/service/run_dispatch.go @@ -105,7 +105,7 @@ func (h *RunHandle) Stream() RunStream { } func (s *Service) SubmitRun(ctx context.Context, req SubmitRunRequest) (RunResult, error) { - if req.ModelGroupKey != "" && s.queue != nil { + if req.ProviderPool && req.ModelGroupKey != "" && s.queue != nil { return s.submitRunQueued(ctx, req) } return s.submitRunDirect(req) @@ -484,6 +484,60 @@ func isProviderAvailable(health string) bool { return h == "available" || h == "healthy" } +// isProviderAdapterInstanceValid is a defensive check in provider-pool candidate +// resolution. It returns false only when the adapter name resolves to a disabled +// exact instance, an ambiguous type route (2+ enabled instances of that type), +// a type route with zero enabled instances, or an unknown/missing adapter key. +// Only enabled exact instances and single-enabled type routes are considered valid. +func isProviderAdapterInstanceValid(rec *edgenode.NodeRecord, adapter string) bool { + if rec == nil { + return true + } + for _, inst := range rec.Adapters.OllamaInstances { + if inst.Name == adapter { + return inst.Enabled + } + } + for _, inst := range rec.Adapters.VllmInstances { + if inst.Name == adapter { + return inst.Enabled + } + } + for _, inst := range rec.Adapters.OpenAICompatInstances { + if inst.Name == adapter { + return inst.Enabled + } + } + if adapter == "cli" { + return rec.Adapters.CLI.Enabled + } + switch adapter { + case "ollama": + // Match edgevalidate.buildAdapterIndex: legacy ollama.Enabled counts as one enabled instance/key. + count := 0 + if rec.Adapters.Ollama.Enabled { + count++ + } + count += len(ollamaEnabledInstances(rec)) + return count == 1 + case "vllm": + count := 0 + if rec.Adapters.Vllm.Enabled { + count++ + } + count += len(vllmEnabledInstances(rec)) + return count == 1 + case "openai_compat": + count := 0 + if rec.Adapters.OpenAICompat.Enabled { + count++ + } + count += len(openAICompatEnabledInstances(rec)) + return count == 1 + } + return false // unknown/custom adapter key: excluded from provider-pool candidates +} + // resolveProviderPoolCandidates builds candidates for a provider-pool request. // It scans connected nodes for providers referenced in the model catalog entry // that matches req.ModelGroupKey, then assembles per-provider candidateNodes @@ -531,6 +585,10 @@ func (s *Service) resolveProviderPoolCandidates(req SubmitRunRequest, store *edg if !providerDispatchable(prov) { continue } + // Defensive: adapter must resolve to an enabled instance on this node. + if !isProviderAdapterInstanceValid(rec, prov.Adapter) { + continue + } // Only available/healthy providers are dispatchable. if !isProviderAvailable(prov.Health) { continue diff --git a/apps/edge/internal/service/service_internal_test.go b/apps/edge/internal/service/service_internal_test.go index 739c230..bdf7dc6 100644 --- a/apps/edge/internal/service/service_internal_test.go +++ b/apps/edge/internal/service/service_internal_test.go @@ -1,126 +1,14 @@ package service import ( - "context" "errors" - "net" - "strings" "sync" "testing" - "time" - toki "git.toki-labs.com/toki/proto-socket/go" - "google.golang.org/protobuf/proto" - - edgeevents "iop/apps/edge/internal/events" edgenode "iop/apps/edge/internal/node" "iop/packages/go/config" - iop "iop/proto/gen/iop" ) -func TestSubmitRunModelQueueUsesRoutePolicyBeforeProviderInstancePolicy(t *testing.T) { - parserMap := toki.ParserMap{ - toki.TypeNameOf(&iop.RunRequest{}): func(b []byte) (proto.Message, error) { - m := &iop.RunRequest{} - return m, proto.Unmarshal(b, m) - }, - } - - edgeConn, nodeConn := net.Pipe() - defer edgeConn.Close() - defer nodeConn.Close() - - edgeClient := toki.NewTcpClient(edgeConn, 0, 0, parserMap) - nodeClient := toki.NewTcpClient(nodeConn, 0, 0, parserMap) - toki.AddListenerTyped[*iop.RunRequest](&nodeClient.Communicator, func(*iop.RunRequest) {}) - - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{NodeID: "prov-node-route-policy", Client: edgeClient}) - - store := edgenode.NewNodeStore() - store.Add(&edgenode.NodeRecord{ - ID: "prov-node-route-policy", - Adapters: config.AdaptersConf{ - OllamaInstances: []config.OllamaInstanceConf{ - { - Name: "ollama-local", - Enabled: true, - Capacity: 1, - MaxQueue: 10, - }, - }, - }, - Runtime: config.RuntimeConf{Concurrency: 1}, - }) - - bus := edgeevents.NewBus() - svc := New(reg, bus) - svc.SetNodeStore(store) - - // run 1: fills capacity (capacity is 1) - res1, err := svc.SubmitRun(context.Background(), SubmitRunRequest{ - ModelGroupKey: "route-policy-group", - Adapter: "ollama-local", - Background: true, - MaxQueue: 1, - }) - if err != nil { - t.Fatalf("run1 error: %v", err) - } - defer res1.Close() - - // run 2: enters queue in a separate goroutine - ctx2, cancel2 := context.WithCancel(context.Background()) - defer cancel2() - errCh2 := make(chan error, 1) - go func() { - res2, err := svc.SubmitRun(ctx2, SubmitRunRequest{ - ModelGroupKey: "route-policy-group", - Adapter: "ollama-local", - Background: true, - MaxQueue: 1, - }) - if err == nil { - res2.Close() - } - errCh2 <- err - }() - - // Wait explicitly until run 2 enters the queue - waitForQueueLen(t, svc.queue, "route-policy-group", 1) - - // run 3: should be rejected immediately because queue is full (max queue = 1, and run2 is in the queue) - // We use a short timeout context to prevent blocking indefinitely if something fails. - ctx3, cancel3 := context.WithTimeout(context.Background(), 2*time.Second) - defer cancel3() - - _, err3 := svc.SubmitRun(ctx3, SubmitRunRequest{ - ModelGroupKey: "route-policy-group", - Adapter: "ollama-local", - Background: true, - MaxQueue: 1, - }) - if err3 == nil { - t.Fatal("expected third run to fail immediately due to queue full") - } - if !strings.Contains(err3.Error(), "queue is full") { - t.Errorf("expected queue is full error, got: %v", err3) - } - - // clean up run2 and verify it exits - cancel2() - select { - case err2 := <-errCh2: - if err2 == nil { - t.Error("expected run2 to fail with context cancelled, got nil") - } else if !errors.Is(err2, context.Canceled) { - t.Errorf("expected run2 to fail with context cancelled, got: %v", err2) - } - case <-time.After(1 * time.Second): - t.Error("timeout waiting for run2 goroutine to exit") - } -} - // TestRefreshProviderCapacityAffectsNextDispatch verifies that a provider // capacity change applied via SetRuntimeConfig is reflected in the candidate // capacity used for the next dispatch admission, since candidates are rebuilt @@ -134,6 +22,11 @@ func TestRefreshProviderCapacityAffectsNextDispatch(t *testing.T) { store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ ID: "node-cap", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: "prov-a", @@ -187,6 +80,11 @@ func TestRuntimeConfigSnapshotConcurrentReplace(t *testing.T) { store.Add(&edgenode.NodeRecord{ ID: "node-refresh", Runtime: config.RuntimeConf{Concurrency: capacity}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, Providers: []config.NodeProviderConf{ { ID: providerID, diff --git a/apps/edge/internal/service/service_test.go b/apps/edge/internal/service/service_test.go index 06d2151..7f4b999 100644 --- a/apps/edge/internal/service/service_test.go +++ b/apps/edge/internal/service/service_test.go @@ -945,142 +945,6 @@ func TestServiceCapabilitiesPreservesProviderSnapshots(t *testing.T) { } } -// TestSubmitRunModelQueueFiltersCandidatesByAdapterTarget verifies that only -// nodes whose adapter config supports the requested adapter/target are used -// as candidates for queued dispatch. -func TestSubmitRunModelQueueFiltersCandidatesByAdapterTarget(t *testing.T) { - parserMap := toki.ParserMap{ - toki.TypeNameOf(&iop.RunRequest{}): func(b []byte) (proto.Message, error) { - m := &iop.RunRequest{} - return m, proto.Unmarshal(b, m) - }, - } - - edgeConn1, nodeConn1 := net.Pipe() - edgeConn2, nodeConn2 := net.Pipe() - defer edgeConn1.Close() - defer nodeConn1.Close() - defer edgeConn2.Close() - defer nodeConn2.Close() - - edgeClient1 := toki.NewTcpClient(edgeConn1, 0, 0, parserMap) - edgeClient2 := toki.NewTcpClient(edgeConn2, 0, 0, parserMap) - nodeClient1 := toki.NewTcpClient(nodeConn1, 0, 0, parserMap) - nodeClient2 := toki.NewTcpClient(nodeConn2, 0, 0, parserMap) - toki.AddListenerTyped[*iop.RunRequest](&nodeClient1.Communicator, func(*iop.RunRequest) {}) - toki.AddListenerTyped[*iop.RunRequest](&nodeClient2.Communicator, func(*iop.RunRequest) {}) - - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{NodeID: "filt-node-1", Client: edgeClient1}) - reg.Register(&edgenode.NodeEntry{NodeID: "filt-node-2", Client: edgeClient2}) - - store := edgenode.NewNodeStore() - // filt-node-1: supports cli/codex (capacity=2 via Concurrency). - store.Add(&edgenode.NodeRecord{ - ID: "filt-node-1", - Adapters: config.AdaptersConf{ - CLI: config.CLIConf{ - Enabled: true, - Profiles: map[string]config.CLIProfileConf{"codex": {Command: "codex"}}, - }, - }, - Runtime: config.RuntimeConf{Concurrency: 2}, - }) - // filt-node-2: supports cli/claude but NOT cli/codex. - store.Add(&edgenode.NodeRecord{ - ID: "filt-node-2", - Adapters: config.AdaptersConf{ - CLI: config.CLIConf{ - Enabled: true, - Profiles: map[string]config.CLIProfileConf{"claude": {Command: "claude"}}, - }, - }, - Runtime: config.RuntimeConf{Concurrency: 2}, - }) - - bus := edgeevents.NewBus() - svc := edgeservice.New(reg, bus) - svc.SetNodeStore(store) - - // Submit 2 runs for cli/codex — both should go to filt-node-1 (cap=2). - var ( - mu sync.Mutex - nodeIDs []string - errs []error - ) - var wg sync.WaitGroup - for i := 0; i < 2; i++ { - wg.Add(1) - go func() { - defer wg.Done() - res, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ - ModelGroupKey: "filt-group", - Adapter: "cli", - Target: "codex", - Background: true, - }) - mu.Lock() - defer mu.Unlock() - if err != nil { - errs = append(errs, err) - return - } - nodeIDs = append(nodeIDs, res.Dispatch().NodeID) - res.Close() - }() - } - wg.Wait() - - for _, err := range errs { - t.Fatalf("SubmitRun error: %v", err) - } - if len(nodeIDs) != 2 { - t.Fatalf("expected 2 dispatches, got %d", len(nodeIDs)) - } - for _, id := range nodeIDs { - if id != "filt-node-1" { - t.Errorf("dispatch to %q: filt-node-2 does not support cli/codex", id) - } - } -} - -// TestSubmitRunModelQueueRejectsExplicitNodeRefWithoutTarget verifies that when -// an explicit NodeRef is given, the node must support the requested adapter/target. -func TestSubmitRunModelQueueRejectsExplicitNodeRefWithoutTarget(t *testing.T) { - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{NodeID: "excl-node-1"}) - - store := edgenode.NewNodeStore() - // excl-node-1: supports cli/claude but NOT cli/codex. - store.Add(&edgenode.NodeRecord{ - ID: "excl-node-1", - Adapters: config.AdaptersConf{ - CLI: config.CLIConf{ - Enabled: true, - Profiles: map[string]config.CLIProfileConf{"claude": {Command: "claude"}}, - }, - }, - }) - - bus := edgeevents.NewBus() - svc := edgeservice.New(reg, bus) - svc.SetNodeStore(store) - - _, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ - NodeRef: "excl-node-1", - ModelGroupKey: "excl-group", - Adapter: "cli", - Target: "codex", - Background: true, - }) - if err == nil { - t.Fatal("expected error: node does not support cli/codex") - } - if !strings.Contains(err.Error(), "does not support") { - t.Errorf("expected 'does not support' in error, got: %v", err) - } -} - // TestSubmitRunModelQueueDispatchesQueuedRunAfterDisconnectToLiveNode verifies // that after a node disconnect, a queued run is dispatched to a remaining live // node once that node's slot becomes available. @@ -1110,24 +974,48 @@ func TestSubmitRunModelQueueDispatchesQueuedRunAfterDisconnectToLiveNode(t *test reg.Register(&edgenode.NodeEntry{NodeID: "dc-node-1", Client: edgeClient1}) reg.Register(&edgenode.NodeEntry{NodeID: "dc-node-2", Client: edgeClient2}) + catalog := []config.ModelCatalogEntry{ + { + ID: "dc-group", + Providers: map[string]string{ + "prov-dc-1": "dc-model", + "prov-dc-2": "dc-model", + }, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ - ID: "dc-node-1", - Runtime: config.RuntimeConf{Concurrency: 1}, + ID: "dc-node-1", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "mock", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-dc-1", Adapter: "mock", Models: []string{"dc-model"}, Health: "available", Capacity: 1}, + }, }) store.Add(&edgenode.NodeRecord{ - ID: "dc-node-2", - Runtime: config.RuntimeConf{Concurrency: 1}, + ID: "dc-node-2", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "mock", Enabled: true, Endpoint: "http://127.0.0.1:8001/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-dc-2", Adapter: "mock", Models: []string{"dc-model"}, Health: "available", Capacity: 1}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) // Submit two background runs to fill both nodes. res1, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "dc-group", - Adapter: "mock", + ProviderPool: true, Background: true, }) if err != nil { @@ -1137,7 +1025,7 @@ func TestSubmitRunModelQueueDispatchesQueuedRunAfterDisconnectToLiveNode(t *test res2, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "dc-group", - Adapter: "mock", + ProviderPool: true, Background: true, }) if err != nil { @@ -1162,7 +1050,7 @@ func TestSubmitRunModelQueueDispatchesQueuedRunAfterDisconnectToLiveNode(t *test defer wg.Done() res3, err3 = svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "dc-group", - Adapter: "mock", + ProviderPool: true, Background: true, }) }() @@ -1217,20 +1105,34 @@ func TestSubmitRunModelQueueContextCancelRemovesQueuedItem(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "ctx-node-1", Client: edgeClient}) + catalog := []config.ModelCatalogEntry{ + { + ID: "ctx-group", + Providers: map[string]string{"prov-ctx-1": "ctx-model"}, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ - ID: "ctx-node-1", - Runtime: config.RuntimeConf{Concurrency: 1}, + ID: "ctx-node-1", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "mock", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-ctx-1", Adapter: "mock", Models: []string{"ctx-model"}, Health: "available", Capacity: 1}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) // Fill the node capacity with run1. res1, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "ctx-group", - Adapter: "mock", + ProviderPool: true, Background: true, }) if err != nil { @@ -1247,7 +1149,7 @@ func TestSubmitRunModelQueueContextCancelRemovesQueuedItem(t *testing.T) { defer wg.Done() _, run2Err = svc.SubmitRun(ctx, edgeservice.SubmitRunRequest{ ModelGroupKey: "ctx-group", - Adapter: "mock", + ProviderPool: true, Background: true, }) }() @@ -1305,21 +1207,45 @@ func TestSubmitRunModelQueueDispatchesAcrossNodes(t *testing.T) { reg.Register(&edgenode.NodeEntry{NodeID: "mq-node-1", Client: edgeClient1}) reg.Register(&edgenode.NodeEntry{NodeID: "mq-node-2", Client: edgeClient2}) + catalog := []config.ModelCatalogEntry{ + { + ID: "test-group", + Providers: map[string]string{ + "prov-mq-1": "served-model", + "prov-mq-2": "served-model", + }, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ - ID: "mq-node-1", - Token: "tok-1", - Runtime: config.RuntimeConf{Concurrency: 1}, + ID: "mq-node-1", + Token: "tok-1", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "mock", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-mq-1", Adapter: "mock", Models: []string{"served-model"}, Health: "available", Capacity: 1}, + }, }) store.Add(&edgenode.NodeRecord{ - ID: "mq-node-2", - Token: "tok-2", - Runtime: config.RuntimeConf{Concurrency: 1}, + ID: "mq-node-2", + Token: "tok-2", + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "mock", Enabled: true, Endpoint: "http://127.0.0.1:8001/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-mq-2", Adapter: "mock", Models: []string{"served-model"}, Health: "available", Capacity: 1}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) var ( mu sync.Mutex @@ -1333,8 +1259,7 @@ func TestSubmitRunModelQueueDispatchesAcrossNodes(t *testing.T) { defer wg.Done() res, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "test-group", - Adapter: "cli", - Target: "codex", + ProviderPool: true, Background: true, }) mu.Lock() @@ -1385,35 +1310,33 @@ func TestSubmitRunModelQueueUsesProviderInstancePolicy(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "prov-node-1", Client: edgeClient}) + catalog := []config.ModelCatalogEntry{ + { + ID: "local-group", + Providers: map[string]string{"prov-local-1": "local-model"}, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ ID: "prov-node-1", Adapters: config.AdaptersConf{ OllamaInstances: []config.OllamaInstanceConf{ - { - Name: "ollama-local", - Enabled: true, - Capacity: 2, - MaxQueue: 3, - }, - { - Name: "ollama-remote", - Enabled: true, - Capacity: 5, - MaxQueue: 10, - }, + {Name: "ollama-local", Enabled: true, BaseURL: "http://127.0.0.1:11434"}, }, }, - Runtime: config.RuntimeConf{Concurrency: 1}, + Providers: []config.NodeProviderConf{ + {ID: "prov-local-1", Adapter: "ollama-local", Models: []string{"local-model"}, Health: "available", Capacity: 2, MaxQueue: 3}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) res1, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "local-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, }) if err != nil { @@ -1423,7 +1346,7 @@ func TestSubmitRunModelQueueUsesProviderInstancePolicy(t *testing.T) { res2, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "local-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, }) if err != nil { @@ -1435,7 +1358,7 @@ func TestSubmitRunModelQueueUsesProviderInstancePolicy(t *testing.T) { defer cancel() _, err3 := svc.SubmitRun(ctx, edgeservice.SubmitRunRequest{ ModelGroupKey: "local-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, }) if err3 == nil { @@ -1446,57 +1369,6 @@ func TestSubmitRunModelQueueUsesProviderInstancePolicy(t *testing.T) { } } -func TestSubmitRunModelQueueRejectsAmbiguousProviderType(t *testing.T) { - reg := edgenode.NewRegistry() - reg.Register(&edgenode.NodeEntry{NodeID: "ambig-node-1"}) - - store := edgenode.NewNodeStore() - store.Add(&edgenode.NodeRecord{ - ID: "ambig-node-1", - Adapters: config.AdaptersConf{ - OllamaInstances: []config.OllamaInstanceConf{ - { - Name: "ollama-local", - Enabled: true, - }, - { - Name: "ollama-remote", - Enabled: true, - }, - }, - }, - }) - - bus := edgeevents.NewBus() - svc := edgeservice.New(reg, bus) - svc.SetNodeStore(store) - - _, err1 := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ - NodeRef: "ambig-node-1", - ModelGroupKey: "ambig-group", - Adapter: "ollama", - Background: true, - }) - if err1 == nil { - t.Fatal("expected error for ambiguous adapter type") - } - if !strings.Contains(err1.Error(), "ambiguous") { - t.Errorf("expected ambiguous error, got: %v", err1) - } - - _, err2 := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ - ModelGroupKey: "ambig-group", - Adapter: "ollama", - Background: true, - }) - if err2 == nil { - t.Fatal("expected error when no candidates support ambiguous adapter type") - } - if !strings.Contains(err2.Error(), "no nodes support") { - t.Errorf("expected no support error, got: %v", err2) - } -} - func TestSubmitRunModelQueueFallsBackToProviderInstancePolicy(t *testing.T) { parserMap := toki.ParserMap{ toki.TypeNameOf(&iop.RunRequest{}): func(b []byte) (proto.Message, error) { @@ -1516,31 +1388,35 @@ func TestSubmitRunModelQueueFallsBackToProviderInstancePolicy(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "prov-node-fallback", Client: edgeClient}) + catalog := []config.ModelCatalogEntry{ + { + ID: "fallback-group", + Providers: map[string]string{"prov-fallback-1": "fallback-model"}, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ ID: "prov-node-fallback", Adapters: config.AdaptersConf{ OllamaInstances: []config.OllamaInstanceConf{ - { - Name: "ollama-local", - Enabled: true, - Capacity: 2, - MaxQueue: 3, - }, + {Name: "ollama-local", Enabled: true, BaseURL: "http://127.0.0.1:11434"}, }, }, - Runtime: config.RuntimeConf{Concurrency: 1}, + Providers: []config.NodeProviderConf{ + // capacity=2, maxQueue=3 come from provider config (provider-pool always uses provider policy). + {ID: "prov-fallback-1", Adapter: "ollama-local", Models: []string{"fallback-model"}, Health: "available", Capacity: 2, MaxQueue: 3}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) res1, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "fallback-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, - MaxQueue: 0, }) if err != nil { t.Fatalf("run1 error: %v", err) @@ -1549,9 +1425,8 @@ func TestSubmitRunModelQueueFallsBackToProviderInstancePolicy(t *testing.T) { res2, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ ModelGroupKey: "fallback-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, - MaxQueue: 0, }) if err != nil { t.Fatalf("run2 error: %v", err) @@ -1562,9 +1437,8 @@ func TestSubmitRunModelQueueFallsBackToProviderInstancePolicy(t *testing.T) { defer cancel() _, err3 := svc.SubmitRun(ctx, edgeservice.SubmitRunRequest{ ModelGroupKey: "fallback-group", - Adapter: "ollama-local", + ProviderPool: true, Background: true, - MaxQueue: 0, }) if err3 == nil { t.Fatal("expected third run to block and timeout (capacity=2 exceeded)") @@ -1593,48 +1467,51 @@ func TestSubmitRunModelQueueUsesRouteQueueTimeout(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "prov-node-timeout", Client: edgeClient}) + catalog := []config.ModelCatalogEntry{ + { + ID: "timeout-group", + Providers: map[string]string{"prov-timeout-1": "timeout-model"}, + }, + } store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ ID: "prov-node-timeout", Adapters: config.AdaptersConf{ OllamaInstances: []config.OllamaInstanceConf{ - { - Name: "ollama-local", - Enabled: true, - Capacity: 1, - MaxQueue: 5, - }, + {Name: "ollama-local", Enabled: true, BaseURL: "http://127.0.0.1:11434"}, }, }, - Runtime: config.RuntimeConf{Concurrency: 1}, + Providers: []config.NodeProviderConf{ + // QueueTimeoutMS=10 comes from provider config; provider-pool ignores request-level QueueTimeoutMS. + {ID: "prov-timeout-1", Adapter: "ollama-local", Models: []string{"timeout-model"}, Health: "available", Capacity: 1, MaxQueue: 5, QueueTimeoutMS: 10}, + }, }) bus := edgeevents.NewBus() svc := edgeservice.New(reg, bus) svc.SetNodeStore(store) + svc.SetModelCatalog(catalog) - // run 1: fills capacity. Also registers QueueTimeoutMS: 10 on the new queue group + // run 1: fills capacity; provider policy sets QueueTimeoutMS=10. res1, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ - ModelGroupKey: "timeout-group", - Adapter: "ollama-local", - Background: true, - QueueTimeoutMS: 10, + ModelGroupKey: "timeout-group", + ProviderPool: true, + Background: true, }) if err != nil { t.Fatalf("run1 error: %v", err) } defer res1.Close() - // run 2: enters queue, using the same QueueTimeoutMS: 10 + // run 2: enters queue; times out after ~10ms per provider policy. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() start := time.Now() _, err2 := svc.SubmitRun(ctx, edgeservice.SubmitRunRequest{ - ModelGroupKey: "timeout-group", - Adapter: "ollama-local", - Background: true, - QueueTimeoutMS: 10, + ModelGroupKey: "timeout-group", + ProviderPool: true, + Background: true, }) elapsed := time.Since(start) @@ -1648,3 +1525,49 @@ func TestSubmitRunModelQueueUsesRouteQueueTimeout(t *testing.T) { t.Errorf("expected timeout to happen quickly, took: %v", elapsed) } } + +// TestSubmitRunLegacyModelGroupKeyGoesDirectNotQueued verifies that a request +// with ModelGroupKey but ProviderPool=false bypasses the queue and goes directly +// to the single registered node. This proves the queue admission gate on +// req.ProviderPool: without ProviderPool=true, ModelGroupKey is decorative only +// and the service falls back to direct dispatch without touching the model catalog. +func TestSubmitRunLegacyModelGroupKeyGoesDirectNotQueued(t *testing.T) { + parserMap := toki.ParserMap{ + toki.TypeNameOf(&iop.RunRequest{}): func(b []byte) (proto.Message, error) { + m := &iop.RunRequest{} + return m, proto.Unmarshal(b, m) + }, + } + + edgeConn, nodeConn := net.Pipe() + defer edgeConn.Close() + defer nodeConn.Close() + + edgeClient := toki.NewTcpClient(edgeConn, 0, 0, parserMap) + nodeClient := toki.NewTcpClient(nodeConn, 0, 0, parserMap) + toki.AddListenerTyped[*iop.RunRequest](&nodeClient.Communicator, func(*iop.RunRequest) {}) + + reg := edgenode.NewRegistry() + reg.Register(&edgenode.NodeEntry{NodeID: "leg-node-1", Client: edgeClient}) + + bus := edgeevents.NewBus() + svc := edgeservice.New(reg, bus) + // No catalog set — if ProviderPool were true, resolveProviderPoolCandidates would + // return "not found in catalog". Since ProviderPool=false, direct dispatch succeeds. + + res, err := svc.SubmitRun(context.Background(), edgeservice.SubmitRunRequest{ + ModelGroupKey: "legacy-group", + Adapter: "cli", + Target: "codex", + Background: true, + // ProviderPool: false (default) + }) + if err != nil { + t.Fatalf("expected direct dispatch to succeed, got: %v", err) + } + defer res.Close() + + if res.Dispatch().NodeID != "leg-node-1" { + t.Errorf("dispatch node: got %q, want leg-node-1", res.Dispatch().NodeID) + } +} diff --git a/apps/edge/internal/service/status_provider.go b/apps/edge/internal/service/status_provider.go index 19ccace..c32392a 100644 --- a/apps/edge/internal/service/status_provider.go +++ b/apps/edge/internal/service/status_provider.go @@ -42,8 +42,12 @@ func (s *Service) ListNodeSnapshots() []NodeSnapshot { } } } - if rec != nil && s.queue != nil { - snap.ProviderSnapshots = s.queue.getSnapshotForNode(entry.NodeID, rec) + if rec != nil { + if s.queue != nil { + snap.ProviderSnapshots = s.queue.getSnapshotForNode(entry.NodeID, rec) + } else if len(rec.Providers) > 0 { + snap.ProviderSnapshots = staticProviderCatalogSnapshots(rec) + } } out = append(out, snap) } @@ -64,6 +68,39 @@ func NodeEntrySnapshot(entry *edgenode.NodeEntry) NodeSnapshot { } } +// staticProviderCatalogSnapshots builds catalog-based ProviderSnapshots with +// zero inflight/queued for nodes whose config has a providers[] catalog but +// whose service has no queue (e.g. queue disabled at startup). +func staticProviderCatalogSnapshots(rec *edgenode.NodeRecord) []*iop.ProviderSnapshot { + concurrencyFallback := positiveOr(rec.Runtime.Concurrency, defaultNodeCapacity) + snaps := make([]*iop.ProviderSnapshot, 0, len(rec.Providers)) + for _, prov := range rec.Providers { + if prov.ID == "" { + continue + } + capVal := prov.Capacity + if capVal <= 0 { + capVal = concurrencyFallback + } + servedModels := make([]string, len(prov.Models)) + copy(servedModels, prov.Models) + lifecycleCaps := make([]string, len(prov.LifecycleCapabilities)) + copy(lifecycleCaps, prov.LifecycleCapabilities) + snaps = append(snaps, &iop.ProviderSnapshot{ + Adapter: prov.Adapter, + Status: "available", + Capacity: int32(capVal), + Id: prov.ID, + Type: prov.Type, + Category: string(prov.Category), + ServedModels: servedModels, + Health: prov.Health, + LifecycleCapabilities: lifecycleCaps, + }) + } + return snaps +} + func (s *Service) ResolveNode(ref string) (*edgenode.NodeEntry, error) { return s.registry.Resolve(ref) } diff --git a/apps/edge/internal/service/status_provider_test.go b/apps/edge/internal/service/status_provider_test.go index 0efb3d2..2da2648 100644 --- a/apps/edge/internal/service/status_provider_test.go +++ b/apps/edge/internal/service/status_provider_test.go @@ -144,7 +144,7 @@ func TestListNodeSnapshotsProviderEmptyAdapterFallbackToProviderID(t *testing.T) // Node has a provider with empty adapter. Queue state is keyed by provider id. store := edgenode.NewNodeStore() rec := &edgenode.NodeRecord{ - ID: "node-e-1", + ID: "node-e-1", Adapters: config.AdaptersConf{ // No OllamaInstances/vllm_instances matching the provider id. // This ensures the fallback path is tested. @@ -224,6 +224,103 @@ func TestListNodeSnapshotsProviderEmptyAdapterFallbackToProviderID(t *testing.T) } } +// TestListNodeSnapshotsProviderCatalogSupersedesAdapterSnapshots verifies the +// catalog-first rule: when a node has both adapter instances (CLI + OpenAICompat) +// and a providers[] catalog, ListNodeSnapshots returns only the catalog snapshots +// with no adapter duplicates. +func TestListNodeSnapshotsProviderCatalogSupersedesAdapterSnapshots(t *testing.T) { + reg := edgenode.NewRegistry() + reg.Register(&edgenode.NodeEntry{NodeID: "node-cs-1", Alias: "node-cs-1"}) + + store := edgenode.NewNodeStore() + rec := &edgenode.NodeRecord{ + ID: "node-cs-1", + Adapters: config.AdaptersConf{ + CLI: config.CLIConf{Enabled: true}, + OpenAICompatInstances: []config.OpenAICompatInstanceConf{ + {Name: "vllm-gpu", Enabled: true}, + }, + }, + Providers: []config.NodeProviderConf{ + {ID: "prov-cat-1", Adapter: "vllm-gpu", Models: []string{"model-x"}, Health: "available", Capacity: 4}, + {ID: "prov-cat-2", Adapter: "vllm-gpu", Models: []string{"model-y"}, Health: "available", Capacity: 2}, + }, + Runtime: config.RuntimeConf{Concurrency: 1}, + } + store.Add(rec) + + bus := edgeevents.NewBus() + svc := New(reg, bus) + svc.SetNodeStore(store) + + snaps := svc.ListNodeSnapshots() + if len(snaps) != 1 { + t.Fatalf("expected 1 node snapshot, got %d", len(snaps)) + } + snap := snaps[0] + + // Only catalog snapshots must be returned — exactly 2 (one per provider). + if len(snap.ProviderSnapshots) != 2 { + t.Fatalf("expected 2 catalog snapshots, got %d (cli/openai_compat adapter duplicates must be absent): %+v", + len(snap.ProviderSnapshots), snap.ProviderSnapshots) + } + ids := map[string]bool{} + for _, p := range snap.ProviderSnapshots { + if p.Id == "" { + t.Errorf("unexpected adapter-only snapshot (empty Id): %+v", p) + } + ids[p.Id] = true + } + if !ids["prov-cat-1"] || !ids["prov-cat-2"] { + t.Errorf("expected catalog IDs prov-cat-1 and prov-cat-2, got: %v", ids) + } +} + +// TestListNodeSnapshotsLegacyAdapterFallbackWhenNoProviders verifies that when +// a node has no providers[] catalog, ListNodeSnapshots returns the legacy adapter +// snapshots (CLI, Ollama, etc.) rather than empty results. +func TestListNodeSnapshotsLegacyAdapterFallbackWhenNoProviders(t *testing.T) { + reg := edgenode.NewRegistry() + reg.Register(&edgenode.NodeEntry{NodeID: "node-legacy-snap", Alias: "node-legacy-snap"}) + + store := edgenode.NewNodeStore() + rec := &edgenode.NodeRecord{ + ID: "node-legacy-snap", + Adapters: config.AdaptersConf{ + CLI: config.CLIConf{Enabled: true}, + OllamaInstances: []config.OllamaInstanceConf{ + {Name: "ollama-local", Enabled: true, Capacity: 3}, + }, + }, + Runtime: config.RuntimeConf{Concurrency: 2}, + // Providers is empty → legacy adapter fallback + } + store.Add(rec) + + bus := edgeevents.NewBus() + svc := New(reg, bus) + svc.SetNodeStore(store) + + snaps := svc.ListNodeSnapshots() + if len(snaps) != 1 { + t.Fatalf("expected 1 node snapshot, got %d", len(snaps)) + } + snap := snaps[0] + + // Must return CLI + ollama-local adapter snapshots. + if len(snap.ProviderSnapshots) != 2 { + t.Fatalf("expected 2 adapter snapshots (cli + ollama-local), got %d: %+v", + len(snap.ProviderSnapshots), snap.ProviderSnapshots) + } + adapters := map[string]bool{} + for _, p := range snap.ProviderSnapshots { + adapters[p.Adapter] = true + } + if !adapters["cli"] || !adapters["ollama-local"] { + t.Errorf("expected cli and ollama-local adapter snapshots, got: %v", adapters) + } +} + func TestListNodeSnapshotsProviderIdDiffersFromAdapterKey(t *testing.T) { reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-p-1", Alias: "node-p-1"})