diff --git a/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/code_review_local_G05_0.log b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/code_review_local_G05_0.log new file mode 100644 index 0000000..49ecf05 --- /dev/null +++ b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/code_review_local_G05_0.log @@ -0,0 +1,157 @@ + + +# Code Review Reference - API + +> **[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 user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed decision in `사용자 리뷰 요청` and stop for code-review. +> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume. +> Follow the ownership table at the bottom of this file for which sections you own. + +## 개요 + +date=2026-06-14 +task=m-provider-availability-capacity-queue-foundation/03_queue_capability_contract, plan=0, tag=API + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-local-G05.md` -> `code_review_local_G05_N.log`, `PLAN-local-G05.md` -> `plan_local_G05_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/`로 이동한다. +4. PASS이고 task group이 `m-provider-availability-capacity-queue-foundation`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [API-1] Runtime Capabilities Queue Fields | [x] | +| [API-2] Provider Adapter Capability Wiring | [x] | + +## 구현 체크리스트 + +- [x] `runtime.Capabilities`에 queue 설정 필드를 추가한다. +- [x] Ollama adapter가 `config.OllamaConf`의 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 보존하고 capabilities에 노출한다. +- [x] vLLM adapter가 `config.VllmConf`의 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 보존하고 capabilities에 노출한다. +- [x] adapter capability tests에 configured queue settings assertion을 추가한다. +- [x] 최종 검증 명령을 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G05_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G05_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-provider-availability-capacity-queue-foundation/03_queue_capability_contract/`를 `agent-task/archive/YYYY/MM/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-provider-availability-capacity-queue-foundation`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-provider-availability-capacity-queue-foundation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-local-G05.md`와 `CODE_REVIEW-local-G05.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로 이동한다. + +## 계획 대비 변경 사항 + +계획과 다른 부분 없이 계획된 사양대로 변경을 모두 적용하였습니다. + +## 주요 설계 결정 + +- `effectiveCapacity` 헬퍼 함수를 도입하여, 각 provider adapter(`ollama`, `vllm`)에서 설정된 capacity 값이 0 이하(지정되지 않음 또는 오류)일 경우, 사전에 정의되어 있던 기본 동시성 제한(Ollama: 4, vLLM: 8)을 폴백으로 취하도록 설계하고 이를 테스트 코드로 보장하였습니다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- `runtime.Capabilities` 필드 추가가 기존 test doubles의 zero value 동작을 깨지 않는지 확인한다. +- Ollama/vLLM의 configured capacity가 기존 default capacity를 0일 때 유지하는지 확인한다. +- queue 실행 동작이 이 subtask에 섞이지 않았는지 확인한다. + +## 검증 결과 + +모든 테스트가 성공적으로 통과하였습니다. + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. + +### API-1 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/runtime +? iop/apps/node/internal/runtime [no test files] +``` + +### API-2 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm +ok iop/apps/node/internal/adapters/ollama 0.011s +ok iop/apps/node/internal/adapters/vllm 0.010s +``` + +### 최종 검증 +```bash +$ go test -count=1 ./apps/node/internal/runtime ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/adapters/ollama 0.009s +ok iop/apps/node/internal/adapters/vllm 0.015s +ok iop/apps/node/internal/adapters 0.007s +``` + +--- + +> **[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 + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS 처리로 `complete.log`를 작성하고 active task 디렉터리를 archive로 이동한다. + +### 리뷰 근거 + +- `runtime.Capabilities`에 queue 관련 필드가 추가되었고 기존 test double은 zero value로 호환된다. +- Ollama/vLLM adapter가 config의 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 struct에 보존하고 capabilities에 노출한다. +- capacity가 0 이하일 때 기존 provider별 기본 동시성(Ollama 4, vLLM 8)으로 fallback된다. +- adapter capability tests가 configured queue 설정과 default capacity fallback을 검증한다. +- 직접 재실행한 최종 검증 명령 결과: + +```bash +$ go test -count=1 ./apps/node/internal/runtime ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/adapters/ollama 0.011s +ok iop/apps/node/internal/adapters/vllm 0.009s +ok iop/apps/node/internal/adapters 0.013s +``` diff --git a/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/complete.log b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/complete.log new file mode 100644 index 0000000..d59bd87 --- /dev/null +++ b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/complete.log @@ -0,0 +1,33 @@ +# Complete - m-provider-availability-capacity-queue-foundation/03_queue_capability_contract + +## 완료 일시 + +2026-06-14 + +## 요약 + +Queue capability contract 구현을 1회 리뷰했고 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_local_G05_0.log` | `code_review_local_G05_0.log` | PASS | runtime/provider adapter capability contract 구현과 focused 검증 통과 | + +## 구현/정리 내용 + +- `runtime.Capabilities`에 `MaxQueue`, `QueueTimeoutMS`, `RequestTimeoutMS`를 추가했다. +- Ollama/vLLM adapter가 provider queue 설정을 보존하고 capabilities에 노출하도록 정리했다. +- Ollama/vLLM capability tests에 configured queue 설정과 default capacity fallback assertion을 추가했다. + +## 최종 검증 + +- `go test -count=1 ./apps/node/internal/runtime ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters` - PASS; runtime은 no test files, Ollama/vLLM/adapters 패키지는 모두 ok. + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/plan_local_G05_0.log b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/plan_local_G05_0.log new file mode 100644 index 0000000..dff15da --- /dev/null +++ b/agent-task/archive/2026/06/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/plan_local_G05_0.log @@ -0,0 +1,195 @@ + + +# Plan - Queue Capability Contract + +## 이 파일을 읽는 구현 에이전트에게 + +구현의 마지막 단계는 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채우는 것이다. 검증을 실행하고 실제 stdout/stderr를 붙여 넣은 뒤 active 파일을 그대로 둔 채 리뷰 준비 완료를 보고한다. 사용자 결정, 사용자 소유 외부 환경, scope conflict 없이는 사용자에게 직접 묻지 말고, 차단 시 review stub의 `사용자 리뷰 요청` 섹션에 근거를 남긴다. `USER_REVIEW.md`, `complete.log`, log/archive 이동은 code-review 전용이다. + +## 배경 + +`capacity-config`의 설정 로딩과 Edge->Node payload 보존은 현재 브랜치에서 작은 작업으로 처리됐다. FIFO queue 본체가 설정값을 안전하게 쓰려면 adapter `Capabilities`가 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 공통 형태로 노출해야 한다. 이 계획은 실행 큐 구현 전에 runtime 계약과 provider adapter capabilities만 좁게 정리한다. + +## 사용자 리뷰 요청 흐름 + +구현 중 사용자-only blocker는 active `CODE_REVIEW-*-G??.md`의 `사용자 리뷰 요청` 섹션에 기록한다. 구현 에이전트는 직접 질문, 채팅 선택지 제시, `request_user_input` 호출을 하지 않는다. code-review가 요청 타당성을 검증하고 실제 `USER_REVIEW.md` 작성을 소유한다. + +## 분석 결과 + +### 읽은 파일 + +- `agent-test/local/rules.md` +- `agent-test/local/node-smoke.md` +- `agent-test/local/platform-common-smoke.md` +- `agent-test/local/edge-smoke.md` +- `agent-roadmap/current.md` +- `agent-roadmap/phase/inference-provider-extension/PHASE.md` +- `agent-roadmap/phase/inference-provider-extension/milestones/provider-availability-capacity-queue-foundation.md` +- `apps/node/internal/runtime/types.go` +- `apps/node/internal/adapters/factory.go` +- `apps/node/internal/adapters/factory_internal_test.go` +- `apps/node/internal/adapters/ollama/ollama.go` +- `apps/node/internal/adapters/ollama/ollama_test.go` +- `apps/node/internal/adapters/vllm/vllm.go` +- `apps/node/internal/adapters/vllm/vllm_test.go` +- `packages/go/config/config.go` +- `packages/go/config/config_test.go` +- `apps/edge/internal/node/mapper.go` +- `apps/edge/internal/node/mapper_test.go` +- `proto/iop/runtime.proto` +- `configs/edge.yaml` + +### 테스트 환경 규칙 + +- `test_env=local`; `agent-test/local/rules.md` 존재 및 읽음. +- 적용 profile: `agent-test/local/node-smoke.md`, `agent-test/local/platform-common-smoke.md`, `agent-test/local/edge-smoke.md`. +- 적용 명령: node는 `go test ./apps/node/...`, platform-common은 `go test ./packages/go/... ./proto/gen/...`, proto 변경 시 `make proto`. +- 이 계획은 node adapter/runtime 계약만 바꾸므로 focused 검증은 `go test -count=1 ./apps/node/internal/runtime ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters`로 둔다. + +### 테스트 커버리지 공백 + +- `runtime.Capabilities`에 queue 설정 필드 추가: 기존 직접 테스트 없음, adapter capability tests에 assertion 추가 필요. +- Ollama/vLLM config -> capabilities 반영: 기존 `TestOllamaCapabilitiesQueryTags`, `TestVllmCapabilitiesQueryModels`, instance key tests가 인접 커버리지다. queue config assertion을 추가한다. +- FIFO admission 동작은 이 계획 범위가 아니며 `04+03_fifo_admission_queue`에서 검증한다. + +### 심볼 참조 + +- renamed/removed symbols: none. +- changed symbols: `runtime.Capabilities` 필드 추가. 참조 위치는 `rg --sort path -n "Capabilities\\{|MaxConcurrency|ProviderStatus" apps/node` 기준으로 adapter tests, `node.go` capabilities handling, test doubles가 있다. + +### 분할 판단 + +- split decision policy 평가 완료. +- 공유 task group: `agent-task/m-provider-availability-capacity-queue-foundation/`. +- `03_queue_capability_contract`: no dependencies. runtime/adapter capability contract only. +- `04+03_fifo_admission_queue`: depends on 03. FIFO admission/release/reject 본체. +- `05+04_queue_observe_snapshot`: depends on 04. queued snapshot 관측 반영. +- API/foundation과 concurrency rollout, observability는 위험과 검증 방식이 달라 split한다. + +### 범위 결정 근거 + +- 포함: `apps/node/internal/runtime/types.go`, Ollama/vLLM adapter capability fields and tests. +- 제외: `apps/node/internal/node` admission queue, reject/release behavior, provider snapshot queued count. 이는 후속 split plan에서 구현한다. +- 제외: Edge service/control-plane snapshot mapper 변경. 현재 proto `ProviderSnapshot`에 `queued`가 이미 있고 Edge는 값을 보존한다. + +### 빌드 등급 + +- `local-G05`: runtime 계약 확장이지만 파일 수와 테스트 범위가 작고 deterministic unit tests로 리뷰 가능하다. + +## 구현 체크리스트 + +- [ ] `runtime.Capabilities`에 queue 설정 필드를 추가한다. +- [ ] Ollama adapter가 `config.OllamaConf`의 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 보존하고 capabilities에 노출한다. +- [ ] vLLM adapter가 `config.VllmConf`의 `capacity`, `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 보존하고 capabilities에 노출한다. +- [ ] adapter capability tests에 configured queue settings assertion을 추가한다. +- [ ] 최종 검증 명령을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [API-1] Runtime Capabilities Queue Fields + +문제: [types.go](/config/workspace/iop/apps/node/internal/runtime/types.go:96)의 `Capabilities`는 `MaxConcurrency`만 있어 `max_queue`, `queue_timeout_ms`, `request_timeout_ms`를 Node admission layer로 전달할 공통 경로가 없다. + +해결 방법: + +Before: +```go +// apps/node/internal/runtime/types.go:96 +type Capabilities struct { + AdapterName string + InstanceKey string + Targets []string + MaxConcurrency int + ProviderStatus ProviderStatus +} +``` + +After: +```go +type Capabilities struct { + AdapterName string + InstanceKey string + Targets []string + MaxConcurrency int + MaxQueue int + QueueTimeoutMS int + RequestTimeoutMS int + ProviderStatus ProviderStatus +} +``` + +수정 파일 및 체크리스트: +- [ ] `apps/node/internal/runtime/types.go`: queue config fields 추가. +- [ ] existing test doubles는 zero value로 유지되는지 확인한다. + +테스트 작성: 별도 runtime package test는 만들지 않는다. 필드 전달은 adapter tests에서 검증한다. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/runtime +``` +예상: no test files 또는 PASS. + +### [API-2] Provider Adapter Capability Wiring + +문제: [ollama.go](/config/workspace/iop/apps/node/internal/adapters/ollama/ollama.go:23)와 [vllm.go](/config/workspace/iop/apps/node/internal/adapters/vllm/vllm.go:25)는 새 queue 설정을 adapter struct에 보존하지 않고, capabilities도 hardcoded capacity만 반환한다. + +해결 방법: + +Before: +```go +// apps/node/internal/adapters/ollama/ollama.go:61 +return runtime.Capabilities{ + AdapterName: Name, + InstanceKey: o.instanceName, + Targets: targets, + MaxConcurrency: 4, + ProviderStatus: runtime.NormalizeProviderStatus(status), +} +``` + +After: +```go +return runtime.Capabilities{ + AdapterName: Name, + InstanceKey: o.instanceName, + Targets: targets, + MaxConcurrency: effectiveCapacity(o.capacity, 4), + MaxQueue: o.maxQueue, + QueueTimeoutMS: o.queueTimeoutMS, + RequestTimeoutMS: o.requestTimeoutMS, + ProviderStatus: runtime.NormalizeProviderStatus(status), +} +``` + +수정 파일 및 체크리스트: +- [ ] `apps/node/internal/adapters/ollama/ollama.go`: struct fields, constructor assignment, capabilities fields. +- [ ] `apps/node/internal/adapters/vllm/vllm.go`: struct fields, constructor assignment, capabilities fields. +- [ ] zero/negative handling은 config load에서 non-negative를 보장하되, adapter는 `capacity<=0`이면 기존 default capacity를 유지한다. + +테스트 작성: +- [ ] `apps/node/internal/adapters/ollama/ollama_test.go`: configured queue values appear in capabilities. +- [ ] `apps/node/internal/adapters/vllm/vllm_test.go`: configured queue values appear in capabilities. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm +``` +예상: PASS. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/node/internal/runtime/types.go` | API-1 | +| `apps/node/internal/adapters/ollama/ollama.go` | API-2 | +| `apps/node/internal/adapters/vllm/vllm.go` | API-2 | +| `apps/node/internal/adapters/ollama/ollama_test.go` | API-2 | +| `apps/node/internal/adapters/vllm/vllm_test.go` | API-2 | + +## 최종 검증 + +```bash +go test -count=1 ./apps/node/internal/runtime ./apps/node/internal/adapters/ollama ./apps/node/internal/adapters/vllm ./apps/node/internal/adapters +``` + +예상: PASS. Go test cache output은 허용하지 않는다. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/CODE_REVIEW-cloud-G06.md b/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/CODE_REVIEW-cloud-G06.md new file mode 100644 index 0000000..769c8e2 --- /dev/null +++ b/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/CODE_REVIEW-cloud-G06.md @@ -0,0 +1,127 @@ + + +# Code Review Reference - 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 user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. + +## 개요 + +date=2026-06-14 +task=m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue, plan=0, tag=REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/provider-availability-capacity-queue-foundation.md` +- Task ids: + - `admission-gate`: Node provider executor가 `in_flight < capacity`일 때만 provider 호출을 시작하고, capacity가 찬 요청은 provider별 FIFO queue에 넣는다. + - `queue-release`: 실행 중 요청이 응답, 실패, 취소, timeout으로 종료되면 `in_flight`를 줄이고 queue의 첫 요청을 실행 슬롯으로 승격한다. + - `queue-reject`: `max_queue` 초과 또는 `queue_timeout` 초과 요청은 명확한 error/rejection reason으로 종료한다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REFACTOR-1] FIFO Admission Manager | [ ] | +| [REFACTOR-2] OnRunRequest Admission Path | [ ] | +| [REFACTOR-3] Queue Rejection Events | [ ] | + +## 구현 체크리스트 + +- [ ] predecessor `03_queue_capability_contract`의 `complete.log`가 active 또는 archive에 있는지 확인한다. +- [ ] `permitManager` 또는 새 admission manager를 FIFO queue semantics로 교체한다. +- [ ] `OnRunRequest`가 capacity 여유가 있으면 즉시 실행하고, capacity가 차면 provider별 FIFO queue에 요청을 대기시킨다. +- [ ] 실행 완료, 실패, 취소, timeout 시 permit을 release하고 queue head를 실행 슬롯으로 승격한다. +- [ ] `max_queue` 초과와 `queue_timeout_ms` 초과는 store status와 RunEvent error에 명확한 reason을 남긴다. +- [ ] foreground/background run 모두 같은 admission path를 사용한다. +- [ ] node unit/integration tests를 FIFO 정책에 맞게 추가/수정한다. +- [ ] 최종 검증 명령을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. + +- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G06_N.log`로 아카이브한다. +- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G06_M.log`로 아카이브한다. +- [ ] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리를 archive로 이동한다. +- [ ] PASS이고 task group이 `m-provider-availability-capacity-queue-foundation`이면 완료 이벤트 메타데이터를 보고하고 roadmap을 직접 수정하지 않는다. + +## 계획 대비 변경 사항 + +_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ + +## 주요 설계 결정 + +_구현 에이전트가 주요 설계 결정 사항을 기록한다._ + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- predecessor `03_queue_capability_contract` 완료 근거가 있는지 확인한다. +- queue admission이 FIFO이며 goroutine/context leak 가능성이 없는지 확인한다. +- foreground/background semantics와 store/event side effects가 테스트와 일치하는지 확인한다. +- full-cycle/e2e 생략이 있으면 사유와 남은 위험이 기록됐는지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +### REFACTOR-1 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/node -run 'Test.*Queue|TestConcurrency' +(output) +``` + +### REFACTOR-2 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/node +(output) +``` + +### REFACTOR-3 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/node -run 'Test.*Reject|Test.*Queue|TestConcurrencyLimit_.*Event' +(output) +``` + +### 최종 검증 +```bash +$ go test -count=1 ./apps/node/internal/node +(output) + +$ go test -count=1 ./apps/node/... +(output) + +$ ./scripts/e2e-smoke.sh +(output) +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** diff --git a/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/PLAN-cloud-G06.md b/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/PLAN-cloud-G06.md new file mode 100644 index 0000000..faf2701 --- /dev/null +++ b/agent-task/m-provider-availability-capacity-queue-foundation/04+03_fifo_admission_queue/PLAN-cloud-G06.md @@ -0,0 +1,243 @@ + + +# Plan - FIFO Admission Queue + +## 이 파일을 읽는 구현 에이전트에게 + +구현의 마지막 단계는 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채우는 것이다. 검증을 실행하고 실제 stdout/stderr를 붙여 넣은 뒤 active 파일을 그대로 둔 채 리뷰 준비 완료를 보고한다. 사용자 결정, 사용자 소유 외부 환경, scope conflict 없이는 사용자에게 직접 묻지 말고, 차단 시 review stub의 `사용자 리뷰 요청` 섹션에 근거를 남긴다. `USER_REVIEW.md`, `complete.log`, log/archive 이동은 code-review 전용이다. + +## 배경 + +현재 Node concurrency gate는 capacity 초과 요청을 즉시 reject한다. roadmap의 `admission-gate`, `queue-release`, `queue-reject`는 이 동작을 provider별 FIFO admission queue로 바꿔야 한다. 이 계획은 `03_queue_capability_contract` 완료 후 Node 실행 경로의 queue 본체를 구현한다. + +## 사용자 리뷰 요청 흐름 + +구현 중 사용자-only blocker는 active `CODE_REVIEW-*-G??.md`의 `사용자 리뷰 요청` 섹션에 기록한다. 구현 에이전트는 직접 질문, 채팅 선택지 제시, `request_user_input` 호출을 하지 않는다. code-review가 요청 타당성을 검증하고 실제 `USER_REVIEW.md` 작성을 소유한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/provider-availability-capacity-queue-foundation.md` +- Task ids: + - `admission-gate`: Node provider executor가 `in_flight < capacity`일 때만 provider 호출을 시작하고, capacity가 찬 요청은 provider별 FIFO queue에 넣는다. + - `queue-release`: 실행 중 요청이 응답, 실패, 취소, timeout으로 종료되면 `in_flight`를 줄이고 queue의 첫 요청을 실행 슬롯으로 승격한다. + - `queue-reject`: `max_queue` 초과 또는 `queue_timeout` 초과 요청은 명확한 error/rejection reason으로 종료한다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-test/local/rules.md` +- `agent-test/local/node-smoke.md` +- `agent-test/local/platform-common-smoke.md` +- `agent-test/local/edge-smoke.md` +- `apps/node/internal/node/node.go` +- `apps/node/internal/node/run_manager.go` +- `apps/node/internal/node/node_test.go` +- `apps/node/internal/node/node_concurrency_integration_test.go` +- `apps/node/internal/runtime/types.go` +- `apps/node/internal/adapters/ollama/ollama.go` +- `apps/node/internal/adapters/vllm/vllm.go` +- `apps/node/internal/adapters/ollama/ollama_test.go` +- `apps/node/internal/adapters/vllm/vllm_test.go` + +### 테스트 환경 규칙 + +- `test_env=local`; `agent-test/local/rules.md` 존재 및 읽음. +- 적용 profile: `agent-test/local/node-smoke.md`. +- 필수 명령: 변경한 node package 또는 `go test ./apps/node/...`. +- 실행 pipeline 변경이므로 focused unit/integration 뒤 `./scripts/e2e-smoke.sh`를 보조 검증으로 실행한다. full-cycle 수동 검증을 수행하지 못하면 review stub에 생략 사유와 남은 위험을 기록한다. + +### 테스트 커버리지 공백 + +- 기존 `TestConcurrencyLimit_*`는 immediate reject 정책을 검증한다. FIFO 대기/승격/timeout/reject 정책에 맞게 새 테스트로 교체해야 한다. +- `node_concurrency_integration_test.go`는 reject event 관측만 검증한다. queue overflow 또는 queue timeout error event 관측으로 갱신해야 한다. +- queue ordering, release after success/failure/cancel/timeout은 현재 커버리지 없음. 새 deterministic tests 필요. + +### 심볼 참조 + +- renamed/removed symbols: none. +- changed symbols 후보: `permitManager.tryAcquire`, `permitManager.release`, `ErrConcurrencyLimitExceeded`. +- call sites: `apps/node/internal/node/node.go` lines 114, 123, 165-166; tests under `apps/node/internal/node/node_test.go` concurrency section; integration test `node_concurrency_integration_test.go`. + +### 분할 판단 + +- split decision policy 평가 완료. +- 이 subtask는 `04+03_fifo_admission_queue`; predecessor `03_queue_capability_contract`가 active sibling으로 있어야 하며 implementation 시작 전 해당 subtask의 `complete.log`가 필요하다. +- 현재 확인 시 active predecessor complete.log는 없음. 구현 에이전트는 `agent-task/m-provider-availability-capacity-queue-foundation/03_queue_capability_contract/complete.log` 또는 archive의 같은 predecessor complete.log가 없으면 시작하지 말고 review stub에 dependency blocker를 기록한다. +- 관측 snapshot은 `05+04_queue_observe_snapshot`로 분리한다. + +### 범위 결정 근거 + +- 포함: Node admission queue, queue release, queue overflow/timeout rejection, tests. +- 제외: Edge service/control-plane changes. Node가 `ProviderSnapshot.Queued` 값을 채우는 것은 후속 05에서 다룬다. +- 제외: priority queue, weighted queue, preemption. roadmap 범위 제외와 일치한다. + +### 빌드 등급 + +- `cloud-G06`: concurrency behavior와 event/store side effects를 바꾸며 기존 reject 정책 테스트를 대체해야 한다. + +## 구현 체크리스트 + +- [ ] predecessor `03_queue_capability_contract`의 `complete.log`가 active 또는 archive에 있는지 확인한다. +- [ ] `permitManager` 또는 새 admission manager를 FIFO queue semantics로 교체한다. +- [ ] `OnRunRequest`가 capacity 여유가 있으면 즉시 실행하고, capacity가 차면 provider별 FIFO queue에 요청을 대기시킨다. +- [ ] 실행 완료, 실패, 취소, timeout 시 permit을 release하고 queue head를 실행 슬롯으로 승격한다. +- [ ] `max_queue` 초과와 `queue_timeout_ms` 초과는 store status와 RunEvent error에 명확한 reason을 남긴다. +- [ ] foreground/background run 모두 같은 admission path를 사용한다. +- [ ] node unit/integration tests를 FIFO 정책에 맞게 추가/수정한다. +- [ ] 최종 검증 명령을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 의존 관계 및 구현 순서 + +- `04+03_fifo_admission_queue`는 디렉터리명 기준으로 `03_queue_capability_contract`의 PASS `complete.log`가 선행 조건이다. +- predecessor가 만족되면 queue manager test를 먼저 작성하고, 그 다음 `OnRunRequest`를 연결한다. + +### [REFACTOR-1] FIFO Admission Manager + +문제: [run_manager.go](/config/workspace/iop/apps/node/internal/node/run_manager.go:55)는 atomic counter 기반 immediate reject만 지원한다. + +해결 방법: + +Before: +```go +// apps/node/internal/node/run_manager.go:66 +func (p *permitManager) tryAcquire() bool { + if p.limit <= 0 { + p.active.Add(1) + return true + } + for { + cur := p.active.Load() + if cur >= p.limit { + return false + } + if p.active.CompareAndSwap(cur, cur+1) { + return true + } + } +} +``` + +After: +```go +type admissionManager struct { + mu sync.Mutex + capacity int + maxQueue int + queueTimeout time.Duration + inFlight int + queue []*admissionRequest +} +``` + +수정 파일 및 체크리스트: +- [ ] `apps/node/internal/node/run_manager.go`: queue manager 타입과 acquire/release/queuedCount API. +- [ ] context cancellation과 queue timeout이 goroutine leak 없이 제거되는지 보장한다. + +테스트 작성: +- [ ] `apps/node/internal/node/node_test.go`: FIFO ordering, max_queue overflow, queue timeout, release after success/failure/cancel/timeout. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/node -run 'Test.*Queue|TestConcurrency' +``` +예상: PASS. + +### [REFACTOR-2] OnRunRequest Admission Path + +문제: [node.go](/config/workspace/iop/apps/node/internal/node/node.go:98)는 foreground/background 모두 capacity 초과 시 `rejectRun`을 호출해 반환한다. queue 대기 상태와 승격 실행 경로가 없다. + +해결 방법: + +Before: +```go +// apps/node/internal/node/node.go:114 +if !n.globalPermit.tryAcquire() { + n.rejectRun(ctx, sess, spec, limit) + return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded) +} +if !adapterPM.tryAcquire() { + n.globalPermit.release() + n.rejectRun(ctx, sess, spec, adapterCap) + return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded) +} +``` + +After: +```go +admission, err := n.admissionFor(spec.Adapter, caps).admit(ctx, request) +if err != nil { + n.rejectRun(ctx, sess, spec, admissionRejectReason(err)) + return err +} +goOrRun(admission) +``` + +수정 파일 및 체크리스트: +- [ ] `apps/node/internal/node/node.go`: resolve 후 store insert 전 admission decision을 수행한다. +- [ ] queued foreground request는 queue admission이 완료될 때까지 기다리고 실행 완료까지 기존 foreground semantics를 유지한다. +- [ ] queued background request는 enqueue 성공 후 반환하고, 승격되면 goroutine에서 실행한다. +- [ ] store status는 queued/running/rejected/completed 흐름이 추적 가능해야 한다. store schema 변경 없이 status string 업데이트로 가능한지 우선 확인한다. + +테스트 작성: +- [ ] foreground second run waits and executes after first release. +- [ ] background queued run returns after enqueue and starts after release. +- [ ] failed/cancelled/timeout run releases next queued request. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/node +``` +예상: PASS. + +### [REFACTOR-3] Queue Rejection Events + +문제: [rejectRun](/config/workspace/iop/apps/node/internal/node/node.go:203)는 concurrency limit exceeded 메시지만 만든다. `max_queue` 초과와 `queue_timeout_ms` 초과 reason이 구분되지 않는다. + +해결 방법: + +Before: +```go +// apps/node/internal/node/node.go:205 +errMsg := fmt.Sprintf("concurrency limit exceeded (limit=%d)", limit) +``` + +After: +```go +errMsg := reason.message() +``` + +수정 파일 및 체크리스트: +- [ ] rejection reason type을 두고 `queue_full`, `queue_timeout`, `concurrency_unavailable`을 구분한다. +- [ ] RunEvent error와 store error에 같은 reason text가 들어가게 한다. +- [ ] 기존 `ErrConcurrencyLimitExceeded` compatibility가 필요한 tests는 새 sentinel 또는 wrapping 정책을 명확히 한다. + +테스트 작성: +- [ ] `node_concurrency_integration_test.go`: queue overflow 또는 timeout error event가 Edge-side listener에 도착한다. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/node -run 'Test.*Reject|Test.*Queue|TestConcurrencyLimit_.*Event' +``` +예상: PASS. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/node/internal/node/run_manager.go` | REFACTOR-1 | +| `apps/node/internal/node/node.go` | REFACTOR-2, REFACTOR-3 | +| `apps/node/internal/node/node_test.go` | REFACTOR-1, REFACTOR-2, REFACTOR-3 | +| `apps/node/internal/node/node_concurrency_integration_test.go` | REFACTOR-3 | + +## 최종 검증 + +```bash +go test -count=1 ./apps/node/internal/node +go test -count=1 ./apps/node/... +./scripts/e2e-smoke.sh +``` + +예상: PASS. Go test cache output은 허용하지 않는다. full-cycle 수동 검증을 수행하지 못하면 `CODE_REVIEW-*-G??.md`에 생략 사유와 남은 위험을 남긴다. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/CODE_REVIEW-local-G05.md b/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/CODE_REVIEW-local-G05.md new file mode 100644 index 0000000..06d6114 --- /dev/null +++ b/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/CODE_REVIEW-local-G05.md @@ -0,0 +1,113 @@ + + +# Code Review Reference - API + +> **[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. + +## 개요 + +date=2026-06-14 +task=m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot, plan=0, tag=API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/provider-availability-capacity-queue-foundation.md` +- Task ids: + - `queue-observe`: Edge가 보는 snapshot에 `capacity`, `in_flight`, `queued`가 포함되어 이후 capacity-aware routing의 입력으로 사용할 수 있다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [API-1] Node Capabilities Queued Count | [ ] | +| [API-2] Edge Snapshot Preservation | [ ] | + +## 구현 체크리스트 + +- [ ] predecessor `04+03_fifo_admission_queue`의 `complete.log`가 active 또는 archive에 있는지 확인한다. +- [ ] Node capabilities command가 admission manager의 actual `in_flight`와 `queued` 값을 읽는다. +- [ ] `NodeCommandResponse.Result["queued"]`와 `ProviderSnapshot.Queued`에 같은 queued count를 넣는다. +- [ ] Edge service/control-plane preservation tests에 queued non-zero fixture를 추가한다. +- [ ] 최종 검증 명령을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. + +- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G05_N.log`로 아카이브한다. +- [ ] active `PLAN-*-G??.md`를 `plan_local_G05_M.log`로 아카이브한다. +- [ ] PASS이면 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리를 archive로 이동한다. +- [ ] PASS이고 task group이 `m-provider-availability-capacity-queue-foundation`이면 완료 이벤트 메타데이터를 보고하고 roadmap을 직접 수정하지 않는다. + +## 계획 대비 변경 사항 + +_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ + +## 주요 설계 결정 + +_구현 에이전트가 주요 설계 결정 사항을 기록한다._ + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- predecessor `04+03_fifo_admission_queue` 완료 근거가 있는지 확인한다. +- Node queued count와 ProviderSnapshot queued count가 같은 source에서 나오는지 확인한다. +- Edge production code 변경 없이 fixture/assertion만으로 보존 경로 검증이 충분한지 확인한다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +### API-1 중간 검증 +```bash +$ go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities' +(output) +``` + +### API-2 중간 검증 +```bash +$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane +(output) +``` + +### 최종 검증 +```bash +$ go test -count=1 ./apps/node/internal/node +(output) + +$ go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane +(output) + +$ go test -count=1 ./apps/node/... ./apps/edge/... +(output) +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** diff --git a/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/PLAN-local-G05.md b/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/PLAN-local-G05.md new file mode 100644 index 0000000..8dfdc8f --- /dev/null +++ b/agent-task/m-provider-availability-capacity-queue-foundation/05+04_queue_observe_snapshot/PLAN-local-G05.md @@ -0,0 +1,172 @@ + + +# Plan - Queue Observation Snapshot + +## 이 파일을 읽는 구현 에이전트에게 + +구현의 마지막 단계는 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채우는 것이다. 검증을 실행하고 실제 stdout/stderr를 붙여 넣은 뒤 active 파일을 그대로 둔 채 리뷰 준비 완료를 보고한다. 사용자 결정, 사용자 소유 외부 환경, scope conflict 없이는 사용자에게 직접 묻지 말고, 차단 시 review stub의 `사용자 리뷰 요청` 섹션에 근거를 남긴다. `USER_REVIEW.md`, `complete.log`, log/archive 이동은 code-review 전용이다. + +## 배경 + +`ProviderSnapshot`에는 이미 `capacity`, `in_flight`, `queued`가 있지만 Node capabilities 응답은 `queued`를 0으로 고정한다. FIFO queue가 들어간 뒤 Edge가 capacity-aware routing 입력으로 쓸 수 있으려면 실제 queued count를 snapshot에 반영해야 한다. 이 계획은 queue 본체 완료 후 관측 표면만 좁게 정리한다. + +## 사용자 리뷰 요청 흐름 + +구현 중 사용자-only blocker는 active `CODE_REVIEW-*-G??.md`의 `사용자 리뷰 요청` 섹션에 기록한다. 구현 에이전트는 직접 질문, 채팅 선택지 제시, `request_user_input` 호출을 하지 않는다. code-review가 요청 타당성을 검증하고 실제 `USER_REVIEW.md` 작성을 소유한다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/provider-availability-capacity-queue-foundation.md` +- Task ids: + - `queue-observe`: Edge가 보는 snapshot에 `capacity`, `in_flight`, `queued`가 포함되어 이후 capacity-aware routing의 입력으로 사용할 수 있다. +- Completion mode: check-on-pass + +## 분석 결과 + +### 읽은 파일 + +- `agent-test/local/rules.md` +- `agent-test/local/node-smoke.md` +- `agent-test/local/edge-smoke.md` +- `apps/node/internal/node/node.go` +- `apps/node/internal/node/run_manager.go` +- `apps/node/internal/node/node_test.go` +- `apps/edge/internal/service/node_command.go` +- `apps/edge/internal/service/status_provider.go` +- `apps/edge/internal/service/service_test.go` +- `apps/edge/internal/controlplane/connector.go` +- `apps/edge/internal/controlplane/connector_test.go` +- `proto/iop/runtime.proto` +- `proto/iop/control.proto` + +### 테스트 환경 규칙 + +- `test_env=local`; `agent-test/local/rules.md` 존재 및 읽음. +- 적용 profile: `agent-test/local/node-smoke.md`, `agent-test/local/edge-smoke.md`. +- Node command response 변경 검증은 `go test -count=1 ./apps/node/internal/node`. +- Edge 보존 경로 검증은 `go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane`. + +### 테스트 커버리지 공백 + +- Node capabilities currently asserts queued `0`; FIFO queue 대기 중 queued `>0` assertion 필요. +- Edge service/control-plane tests already preserve provider snapshot values; queued non-zero fixture로 회귀 보강이 필요하다. + +### 심볼 참조 + +- renamed/removed symbols: none. +- changed fields: `ProviderSnapshot.Queued`, command result `"queued"`. +- call sites found by `rg --sort path -n "ProviderSnapshots|queued|ProviderSnapshot" apps/node apps/edge proto/iop`. + +### 분할 판단 + +- split decision policy 평가 완료. +- 이 subtask는 `05+04_queue_observe_snapshot`; predecessor `04+03_fifo_admission_queue` PASS `complete.log`가 필요하다. +- active predecessor complete.log는 현재 없음. 구현 시작 전 active/archive predecessor completion을 확인한다. + +### 범위 결정 근거 + +- 포함: Node capabilities queued count, Edge snapshot preservation tests. +- 제외: FIFO admission/release/reject 본체. predecessor 04에서 완료되어야 한다. +- 제외: capacity-aware routing. 이 milestone의 범위 제외에 해당한다. + +### 빌드 등급 + +- `local-G05`: predecessor queue manager 위에 관측 값을 연결하는 좁은 change이며 focused tests로 검증 가능하다. + +## 구현 체크리스트 + +- [ ] predecessor `04+03_fifo_admission_queue`의 `complete.log`가 active 또는 archive에 있는지 확인한다. +- [ ] Node capabilities command가 admission manager의 actual `in_flight`와 `queued` 값을 읽는다. +- [ ] `NodeCommandResponse.Result["queued"]`와 `ProviderSnapshot.Queued`에 같은 queued count를 넣는다. +- [ ] Edge service/control-plane preservation tests에 queued non-zero fixture를 추가한다. +- [ ] 최종 검증 명령을 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 의존 관계 및 구현 순서 + +- `05+04_queue_observe_snapshot`는 디렉터리명 기준으로 `04+03_fifo_admission_queue`의 PASS `complete.log`가 선행 조건이다. + +### [API-1] Node Capabilities Queued Count + +문제: [node.go](/config/workspace/iop/apps/node/internal/node/node.go:344)는 capabilities result의 `queued`를 `"0"`으로 고정하고 [node.go](/config/workspace/iop/apps/node/internal/node/node.go:358)의 `ProviderSnapshot.Queued`도 0으로 고정한다. + +해결 방법: + +Before: +```go +// apps/node/internal/node/node.go:350 +"capacity": strconv.Itoa(caps.MaxConcurrency), +"in_flight": strconv.Itoa(inFlight), +"queued": "0", +``` + +After: +```go +stats := n.admissionStats(req.GetAdapter()) +"capacity": strconv.Itoa(stats.Capacity), +"in_flight": strconv.Itoa(stats.InFlight), +"queued": strconv.Itoa(stats.Queued), +``` + +수정 파일 및 체크리스트: +- [ ] `apps/node/internal/node/node.go`: admission stats를 capabilities response에 반영. +- [ ] `apps/node/internal/node/run_manager.go`: queued count accessor가 없다면 추가. + +테스트 작성: +- [ ] `apps/node/internal/node/node_test.go`: capacity full 상태에서 request가 queued일 때 capabilities `queued=1`, snapshot `Queued=1` 검증. + +중간 검증: +```bash +go test -count=1 ./apps/node/internal/node -run 'TestOnCommandRequest_Capabilities' +``` +예상: PASS. + +### [API-2] Edge Snapshot Preservation + +문제: Edge는 provider snapshots를 보존하지만 queued non-zero fixture가 약하면 Node 관측 값이 Edge/Control Plane 표면까지 유지되는지 회귀 검증이 약하다. + +해결 방법: + +Before: +```go +// apps/edge/internal/service/service_test.go fixture +Queued: 0, +``` + +After: +```go +Queued: 3, +``` + +수정 파일 및 체크리스트: +- [ ] `apps/edge/internal/service/service_test.go`: queued non-zero preservation assertion. +- [ ] `apps/edge/internal/controlplane/connector_test.go`: queued non-zero preservation assertion. +- [ ] 소스 production code는 기존 mapper가 값을 보존하면 변경하지 않는다. + +테스트 작성: 위 edge tests fixture/assertion 갱신. + +중간 검증: +```bash +go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane +``` +예상: PASS. + +## 수정 파일 요약 + +| 파일 | 항목 | +|------|------| +| `apps/node/internal/node/node.go` | API-1 | +| `apps/node/internal/node/run_manager.go` | API-1 | +| `apps/node/internal/node/node_test.go` | API-1 | +| `apps/edge/internal/service/service_test.go` | API-2 | +| `apps/edge/internal/controlplane/connector_test.go` | API-2 | + +## 최종 검증 + +```bash +go test -count=1 ./apps/node/internal/node +go test -count=1 ./apps/edge/internal/service ./apps/edge/internal/controlplane +go test -count=1 ./apps/node/... ./apps/edge/... +``` + +예상: PASS. Go test cache output은 허용하지 않는다. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다. diff --git a/apps/edge/internal/node/mapper.go b/apps/edge/internal/node/mapper.go index 04fabd8..0aee324 100644 --- a/apps/edge/internal/node/mapper.go +++ b/apps/edge/internal/node/mapper.go @@ -44,8 +44,12 @@ func BuildConfigPayload(rec *NodeRecord) (*iop.NodeConfigPayload, error) { Name: inst.Name, Config: &iop.AdapterConfig_Ollama{ Ollama: &iop.OllamaAdapterConfig{ - BaseUrl: inst.BaseURL, - ContextSize: int32(inst.ContextSize), + BaseUrl: inst.BaseURL, + ContextSize: int32(inst.ContextSize), + Capacity: int32(inst.Capacity), + MaxQueue: int32(inst.MaxQueue), + QueueTimeoutMs: int32(inst.QueueTimeoutMS), + RequestTimeoutMs: int32(inst.RequestTimeoutMS), }, }, }) @@ -65,7 +69,13 @@ func BuildConfigPayload(rec *NodeRecord) (*iop.NodeConfigPayload, error) { Enabled: true, Name: inst.Name, Config: &iop.AdapterConfig_Vllm{ - Vllm: &iop.VllmAdapterConfig{Endpoint: inst.Endpoint}, + Vllm: &iop.VllmAdapterConfig{ + Endpoint: inst.Endpoint, + Capacity: int32(inst.Capacity), + MaxQueue: int32(inst.MaxQueue), + QueueTimeoutMs: int32(inst.QueueTimeoutMS), + RequestTimeoutMs: int32(inst.RequestTimeoutMS), + }, }, }) } diff --git a/apps/edge/internal/node/mapper_test.go b/apps/edge/internal/node/mapper_test.go index c4c87a7..3531e39 100644 --- a/apps/edge/internal/node/mapper_test.go +++ b/apps/edge/internal/node/mapper_test.go @@ -18,7 +18,16 @@ func TestBuildConfigPayload_OllamaEnabled(t *testing.T) { Token: "token", Adapters: config.AdaptersConf{ OllamaInstances: []config.OllamaInstanceConf{ - {Name: "ollama", Enabled: true, BaseURL: "http://localhost:11434", ContextSize: 262144}, + { + Name: "ollama", + Enabled: true, + BaseURL: "http://localhost:11434", + ContextSize: 262144, + Capacity: 3, + MaxQueue: 7, + QueueTimeoutMS: 1500, + RequestTimeoutMS: 30000, + }, }, }, Runtime: config.RuntimeConf{ @@ -61,6 +70,9 @@ func TestBuildConfigPayload_OllamaEnabled(t *testing.T) { if ollamaCfg.ContextSize != 262144 { t.Errorf("expected ollama context size %d, got %d", 262144, ollamaCfg.ContextSize) } + if ollamaCfg.Capacity != 3 || ollamaCfg.MaxQueue != 7 || ollamaCfg.QueueTimeoutMs != 1500 || ollamaCfg.RequestTimeoutMs != 30000 { + t.Errorf("unexpected ollama queue config: %+v", ollamaCfg) + } } } if !mockFound { @@ -251,8 +263,8 @@ func TestBuildConfigPayload_MultiOllamaInstances(t *testing.T) { Token: "token", Adapters: config.AdaptersConf{ OllamaInstances: []config.OllamaInstanceConf{ - {Name: "local", Enabled: true, BaseURL: "http://127.0.0.1:11434", ContextSize: 131072}, - {Name: "dgx", Enabled: true, BaseURL: "http://192.168.0.91:11434", ContextSize: 262144}, + {Name: "local", Enabled: true, BaseURL: "http://127.0.0.1:11434", ContextSize: 131072, Capacity: 2, MaxQueue: 4, QueueTimeoutMS: 1000, RequestTimeoutMS: 20000}, + {Name: "dgx", Enabled: true, BaseURL: "http://192.168.0.91:11434", ContextSize: 262144, Capacity: 6, MaxQueue: 12, QueueTimeoutMS: 2000, RequestTimeoutMS: 60000}, }, }, Runtime: config.RuntimeConf{WorkspaceRoot: "/tmp/ws"}, @@ -275,6 +287,16 @@ func TestBuildConfigPayload_MultiOllamaInstances(t *testing.T) { names := map[string]bool{} for _, a := range ollamaAdapters { names[a.Name] = true + switch a.Name { + case "local": + if a.GetOllama().GetCapacity() != 2 || a.GetOllama().GetMaxQueue() != 4 || a.GetOllama().GetQueueTimeoutMs() != 1000 || a.GetOllama().GetRequestTimeoutMs() != 20000 { + t.Errorf("local queue config mismatch: %+v", a.GetOllama()) + } + case "dgx": + if a.GetOllama().GetCapacity() != 6 || a.GetOllama().GetMaxQueue() != 12 || a.GetOllama().GetQueueTimeoutMs() != 2000 || a.GetOllama().GetRequestTimeoutMs() != 60000 { + t.Errorf("dgx queue config mismatch: %+v", a.GetOllama()) + } + } } if !names["local"] || !names["dgx"] { t.Errorf("expected instance names local and dgx, got %v", names) @@ -288,8 +310,8 @@ func TestBuildConfigPayload_MultiVllmInstances(t *testing.T) { Token: "token", Adapters: config.AdaptersConf{ VllmInstances: []config.VllmInstanceConf{ - {Name: "a100", Enabled: true, Endpoint: "http://10.0.0.5:8000"}, - {Name: "h100", Enabled: true, Endpoint: "http://10.0.0.6:8000"}, + {Name: "a100", Enabled: true, Endpoint: "http://10.0.0.5:8000", Capacity: 4, MaxQueue: 10, QueueTimeoutMS: 1500, RequestTimeoutMS: 45000}, + {Name: "h100", Enabled: true, Endpoint: "http://10.0.0.6:8000", Capacity: 8, MaxQueue: 16, QueueTimeoutMS: 2500, RequestTimeoutMS: 90000}, }, }, Runtime: config.RuntimeConf{WorkspaceRoot: "/tmp/ws"}, @@ -315,6 +337,16 @@ func TestBuildConfigPayload_MultiVllmInstances(t *testing.T) { if a.GetVllm() == nil { t.Errorf("instance %q: expected vllm typed config", a.Name) } + switch a.Name { + case "a100": + if a.GetVllm().GetCapacity() != 4 || a.GetVllm().GetMaxQueue() != 10 || a.GetVllm().GetQueueTimeoutMs() != 1500 || a.GetVllm().GetRequestTimeoutMs() != 45000 { + t.Errorf("a100 queue config mismatch: %+v", a.GetVllm()) + } + case "h100": + if a.GetVllm().GetCapacity() != 8 || a.GetVllm().GetMaxQueue() != 16 || a.GetVllm().GetQueueTimeoutMs() != 2500 || a.GetVllm().GetRequestTimeoutMs() != 90000 { + t.Errorf("h100 queue config mismatch: %+v", a.GetVllm()) + } + } } if !names["a100"] || !names["h100"] { t.Errorf("expected instance names a100 and h100, got %v", names) @@ -329,7 +361,7 @@ func TestBuildConfigPayload_LegacyOllamaViaInstances(t *testing.T) { Adapters: config.AdaptersConf{ // Simulate what LoadEdge produces after legacy normalisation. OllamaInstances: []config.OllamaInstanceConf{ - {Name: "ollama", Enabled: true, BaseURL: "http://localhost:11434", ContextSize: 4096}, + {Name: "ollama", Enabled: true, BaseURL: "http://localhost:11434", ContextSize: 4096, Capacity: 5, MaxQueue: 9, QueueTimeoutMS: 1700, RequestTimeoutMS: 55000}, }, }, Runtime: config.RuntimeConf{WorkspaceRoot: "/tmp/ws"}, @@ -356,6 +388,9 @@ func TestBuildConfigPayload_LegacyOllamaViaInstances(t *testing.T) { if found.GetOllama().GetBaseUrl() != "http://localhost:11434" { t.Errorf("unexpected base_url: %q", found.GetOllama().GetBaseUrl()) } + if found.GetOllama().GetCapacity() != 5 || found.GetOllama().GetMaxQueue() != 9 || found.GetOllama().GetQueueTimeoutMs() != 1700 || found.GetOllama().GetRequestTimeoutMs() != 55000 { + t.Errorf("unexpected queue config: %+v", found.GetOllama()) + } } func TestBuildConfigPayload_DuplicateOllamaNameError(t *testing.T) { diff --git a/apps/node/internal/adapters/factory.go b/apps/node/internal/adapters/factory.go index 65d2d40..4a43483 100644 --- a/apps/node/internal/adapters/factory.go +++ b/apps/node/internal/adapters/factory.go @@ -60,14 +60,25 @@ func BuildFromPayload(payload *iop.NodeConfigPayload, logger *zap.Logger) (*Regi func ollamaConfFromProto(m *iop.OllamaAdapterConfig) config.OllamaConf { return config.OllamaConf{ - Enabled: true, - BaseURL: m.GetBaseUrl(), - ContextSize: int(m.GetContextSize()), + Enabled: true, + BaseURL: m.GetBaseUrl(), + ContextSize: int(m.GetContextSize()), + Capacity: int(m.GetCapacity()), + MaxQueue: int(m.GetMaxQueue()), + QueueTimeoutMS: int(m.GetQueueTimeoutMs()), + RequestTimeoutMS: int(m.GetRequestTimeoutMs()), } } func vllmConfFromProto(m *iop.VllmAdapterConfig) config.VllmConf { - return config.VllmConf{Enabled: true, Endpoint: m.GetEndpoint()} + return config.VllmConf{ + Enabled: true, + Endpoint: m.GetEndpoint(), + Capacity: int(m.GetCapacity()), + MaxQueue: int(m.GetMaxQueue()), + QueueTimeoutMS: int(m.GetQueueTimeoutMs()), + RequestTimeoutMS: int(m.GetRequestTimeoutMs()), + } } func cliConfFromProto(m *iop.CLIAdapterConfig) config.CLIConf { @@ -120,6 +131,18 @@ func ollamaConfFromStruct(s *structpb.Struct) config.OllamaConf { } else if v, ok := intSetting(m, "num_ctx"); ok { cfg.ContextSize = v } + 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 } @@ -151,6 +174,18 @@ func vllmConfFromStruct(s *structpb.Struct) config.VllmConf { if v, ok := m["endpoint"].(string); ok { cfg.Endpoint = v } + 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 } diff --git a/apps/node/internal/adapters/factory_internal_test.go b/apps/node/internal/adapters/factory_internal_test.go index 5a7e533..1116433 100644 --- a/apps/node/internal/adapters/factory_internal_test.go +++ b/apps/node/internal/adapters/factory_internal_test.go @@ -211,8 +211,12 @@ func TestCLIConfFromStruct_OpencodeSSEModeLegacy(t *testing.T) { func TestOllamaConfFromStruct_LegacyPayload(t *testing.T) { st, err := structpb.NewStruct(map[string]any{ - "base_url": "http://legacy:11434", - "context_size": float64(262144), + "base_url": "http://legacy:11434", + "context_size": float64(262144), + "capacity": float64(3), + "max_queue": float64(8), + "queue_timeout_ms": float64(1500), + "request_timeout_ms": float64(30000), }) if err != nil { t.Fatalf("structpb: %v", err) @@ -227,10 +231,19 @@ func TestOllamaConfFromStruct_LegacyPayload(t *testing.T) { if cfg.ContextSize != 262144 { t.Fatalf("context_size: got %d", cfg.ContextSize) } + if cfg.Capacity != 3 || cfg.MaxQueue != 8 || cfg.QueueTimeoutMS != 1500 || cfg.RequestTimeoutMS != 30000 { + t.Fatalf("queue config: got %+v", cfg) + } } func TestVllmConfFromStruct_LegacyPayload(t *testing.T) { - st, err := structpb.NewStruct(map[string]any{"endpoint": "http://legacy:8000"}) + st, err := structpb.NewStruct(map[string]any{ + "endpoint": "http://legacy:8000", + "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) } @@ -241,6 +254,40 @@ func TestVllmConfFromStruct_LegacyPayload(t *testing.T) { if cfg.Endpoint != "http://legacy:8000" { t.Fatalf("endpoint: got %q", cfg.Endpoint) } + 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", + ContextSize: 8192, + Capacity: 3, + MaxQueue: 8, + QueueTimeoutMs: 1500, + RequestTimeoutMs: 30000, + }) + if ollamaCfg.BaseURL != "http://typed:11434" || ollamaCfg.ContextSize != 8192 { + t.Fatalf("ollama config: got %+v", ollamaCfg) + } + if ollamaCfg.Capacity != 3 || ollamaCfg.MaxQueue != 8 || ollamaCfg.QueueTimeoutMS != 1500 || ollamaCfg.RequestTimeoutMS != 30000 { + t.Fatalf("ollama queue config: got %+v", ollamaCfg) + } + + vllmCfg := vllmConfFromProto(&iop.VllmAdapterConfig{ + Endpoint: "http://typed:8000", + Capacity: 4, + MaxQueue: 10, + QueueTimeoutMs: 2000, + RequestTimeoutMs: 45000, + }) + if vllmCfg.Endpoint != "http://typed:8000" { + t.Fatalf("vllm config: got %+v", vllmCfg) + } + if vllmCfg.Capacity != 4 || vllmCfg.MaxQueue != 10 || vllmCfg.QueueTimeoutMS != 2000 || vllmCfg.RequestTimeoutMS != 45000 { + t.Fatalf("vllm queue config: got %+v", vllmCfg) + } } func TestCLIConfFromStruct_LegacyPayload(t *testing.T) { diff --git a/apps/node/internal/adapters/ollama/ollama.go b/apps/node/internal/adapters/ollama/ollama.go index e6ff7bd..01f0ac1 100644 --- a/apps/node/internal/adapters/ollama/ollama.go +++ b/apps/node/internal/adapters/ollama/ollama.go @@ -21,11 +21,15 @@ import ( const Name = "ollama" type Ollama struct { - instanceName string - baseURL string - contextSize int - client *http.Client - logger *zap.Logger + instanceName string + baseURL string + contextSize int + capacity int + maxQueue int + queueTimeoutMS int + requestTimeoutMS int + client *http.Client + logger *zap.Logger } // New creates an Ollama adapter. The optional instanceName is the registry @@ -40,11 +44,15 @@ func New(cfg config.OllamaConf, logger *zap.Logger, instanceName ...string) *Oll name = instanceName[0] } return &Ollama{ - instanceName: name, - baseURL: baseURL, - contextSize: cfg.ContextSize, - client: &http.Client{}, - logger: logger, + instanceName: name, + baseURL: baseURL, + contextSize: cfg.ContextSize, + capacity: cfg.Capacity, + maxQueue: cfg.MaxQueue, + queueTimeoutMS: cfg.QueueTimeoutMS, + requestTimeoutMS: cfg.RequestTimeoutMS, + client: &http.Client{}, + logger: logger, } } @@ -59,14 +67,24 @@ func (o *Ollama) Capabilities(ctx context.Context) (runtime.Capabilities, error) status = runtime.ProviderStatusUnavailable } return runtime.Capabilities{ - AdapterName: Name, - InstanceKey: o.instanceName, - Targets: targets, - MaxConcurrency: 4, - ProviderStatus: runtime.NormalizeProviderStatus(status), + AdapterName: Name, + InstanceKey: o.instanceName, + Targets: targets, + MaxConcurrency: effectiveCapacity(o.capacity, 4), + MaxQueue: o.maxQueue, + QueueTimeoutMS: o.queueTimeoutMS, + RequestTimeoutMS: o.requestTimeoutMS, + ProviderStatus: runtime.NormalizeProviderStatus(status), }, nil } +func effectiveCapacity(capacity, defaultVal int) int { + if capacity <= 0 { + return defaultVal + } + return capacity +} + func (o *Ollama) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { model := strings.TrimSpace(spec.Target) if model == "" { diff --git a/apps/node/internal/adapters/ollama/ollama_test.go b/apps/node/internal/adapters/ollama/ollama_test.go index d663e08..50440c5 100644 --- a/apps/node/internal/adapters/ollama/ollama_test.go +++ b/apps/node/internal/adapters/ollama/ollama_test.go @@ -197,7 +197,13 @@ func TestOllamaCapabilitiesQueryTags(t *testing.T) { })) defer server.Close() - adapter := New(config.OllamaConf{BaseURL: server.URL}, zap.NewNop()) + adapter := New(config.OllamaConf{ + BaseURL: server.URL, + Capacity: 5, + MaxQueue: 10, + QueueTimeoutMS: 15000, + RequestTimeoutMS: 30000, + }, zap.NewNop()) caps, err := adapter.Capabilities(context.Background()) if err != nil { t.Fatalf("Capabilities failed: %v", err) @@ -206,6 +212,28 @@ func TestOllamaCapabilitiesQueryTags(t *testing.T) { if got != "llama-a,llama-b" { t.Fatalf("targets: got %q", got) } + if caps.MaxConcurrency != 5 { + t.Fatalf("expected MaxConcurrency 5, got %d", caps.MaxConcurrency) + } + if caps.MaxQueue != 10 { + t.Fatalf("expected MaxQueue 10, got %d", caps.MaxQueue) + } + if caps.QueueTimeoutMS != 15000 { + t.Fatalf("expected QueueTimeoutMS 15000, got %d", caps.QueueTimeoutMS) + } + if caps.RequestTimeoutMS != 30000 { + t.Fatalf("expected RequestTimeoutMS 30000, got %d", caps.RequestTimeoutMS) + } + + // Verify default capacity fallback + adapterDefault := New(config.OllamaConf{BaseURL: server.URL}, zap.NewNop()) + capsDefault, err := adapterDefault.Capabilities(context.Background()) + if err != nil { + t.Fatalf("Capabilities failed: %v", err) + } + if capsDefault.MaxConcurrency != 4 { + t.Fatalf("expected default MaxConcurrency 4, got %d", capsDefault.MaxConcurrency) + } } func TestOllamaHandleCommandPassesNativeAPI(t *testing.T) { diff --git a/apps/node/internal/adapters/vllm/vllm.go b/apps/node/internal/adapters/vllm/vllm.go index 5312041..30fb119 100644 --- a/apps/node/internal/adapters/vllm/vllm.go +++ b/apps/node/internal/adapters/vllm/vllm.go @@ -23,10 +23,14 @@ import ( const Name = "vllm" type Vllm struct { - instanceName string - endpoint string - client *http.Client - logger *zap.Logger + instanceName string + endpoint string + capacity int + maxQueue int + queueTimeoutMS int + requestTimeoutMS int + client *http.Client + logger *zap.Logger } // New creates a vLLM adapter. The optional instanceName is the registry @@ -38,10 +42,14 @@ func New(cfg config.VllmConf, logger *zap.Logger, instanceName ...string) *Vllm name = instanceName[0] } return &Vllm{ - instanceName: name, - endpoint: endpoint, - client: &http.Client{}, - logger: logger, + instanceName: name, + endpoint: endpoint, + capacity: cfg.Capacity, + maxQueue: cfg.MaxQueue, + queueTimeoutMS: cfg.QueueTimeoutMS, + requestTimeoutMS: cfg.RequestTimeoutMS, + client: &http.Client{}, + logger: logger, } } @@ -56,14 +64,24 @@ func (v *Vllm) Capabilities(ctx context.Context) (runtime.Capabilities, error) { status = runtime.ProviderStatusUnavailable } return runtime.Capabilities{ - AdapterName: Name, - InstanceKey: v.instanceName, - Targets: targets, - MaxConcurrency: 8, - ProviderStatus: runtime.NormalizeProviderStatus(status), + AdapterName: Name, + InstanceKey: v.instanceName, + Targets: targets, + MaxConcurrency: effectiveCapacity(v.capacity, 8), + MaxQueue: v.maxQueue, + QueueTimeoutMS: v.queueTimeoutMS, + RequestTimeoutMS: v.requestTimeoutMS, + ProviderStatus: runtime.NormalizeProviderStatus(status), }, nil } +func effectiveCapacity(capacity, defaultVal int) int { + if capacity <= 0 { + return defaultVal + } + return capacity +} + func (v *Vllm) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error { if v.endpoint == "" { return fmt.Errorf("vllm adapter: endpoint is required") diff --git a/apps/node/internal/adapters/vllm/vllm_test.go b/apps/node/internal/adapters/vllm/vllm_test.go index 9375a34..f4c3e1a 100644 --- a/apps/node/internal/adapters/vllm/vllm_test.go +++ b/apps/node/internal/adapters/vllm/vllm_test.go @@ -43,7 +43,13 @@ func TestVllmCapabilitiesQueryModels(t *testing.T) { })) defer server.Close() - adapter := New(config.VllmConf{Endpoint: server.URL}, zap.NewNop()) + adapter := New(config.VllmConf{ + 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) @@ -51,6 +57,28 @@ func TestVllmCapabilitiesQueryModels(t *testing.T) { 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 { + t.Fatalf("expected MaxQueue 20, got %d", caps.MaxQueue) + } + if caps.QueueTimeoutMS != 5000 { + t.Fatalf("expected QueueTimeoutMS 5000, got %d", caps.QueueTimeoutMS) + } + if caps.RequestTimeoutMS != 10000 { + t.Fatalf("expected RequestTimeoutMS 10000, got %d", caps.RequestTimeoutMS) + } + + // Verify default capacity fallback + adapterDefault := New(config.VllmConf{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 TestVllmExecuteStreamsDeltas(t *testing.T) { diff --git a/apps/node/internal/runtime/types.go b/apps/node/internal/runtime/types.go index 268ce42..77b0dfd 100644 --- a/apps/node/internal/runtime/types.go +++ b/apps/node/internal/runtime/types.go @@ -95,11 +95,14 @@ func NormalizeProviderStatus(status ProviderStatus) ProviderStatus { // Capabilities describes what an Adapter can do. type Capabilities struct { - AdapterName string - InstanceKey string // stable registry instance key; empty for single-instance adapters - Targets []string - MaxConcurrency int - ProviderStatus ProviderStatus + AdapterName string + InstanceKey string // stable registry instance key; empty for single-instance adapters + Targets []string + MaxConcurrency int + MaxQueue int + QueueTimeoutMS int + RequestTimeoutMS int + ProviderStatus ProviderStatus } // RunRequest is the node-domain representation of an incoming run request. diff --git a/configs/edge.yaml b/configs/edge.yaml index 07fd5dd..aa3c56d 100644 --- a/configs/edge.yaml +++ b/configs/edge.yaml @@ -96,9 +96,17 @@ nodes: enabled: false base_url: "" context_size: 0 + capacity: 0 + max_queue: 0 + queue_timeout_ms: 0 + request_timeout_ms: 0 vllm: enabled: false endpoint: "" + capacity: 0 + max_queue: 0 + queue_timeout_ms: 0 + request_timeout_ms: 0 cli: enabled: true profiles: @@ -150,14 +158,26 @@ nodes: # enabled: true # base_url: "http://127.0.0.1:11434" # context_size: 131072 + # capacity: 4 + # max_queue: 16 + # queue_timeout_ms: 30000 + # request_timeout_ms: 300000 # - name: "ollama-dgx" # enabled: true # base_url: "http://192.168.0.91:11434" # context_size: 262144 + # capacity: 8 + # max_queue: 32 + # queue_timeout_ms: 30000 + # request_timeout_ms: 300000 # vllm_instances: # - name: "vllm-a100" # enabled: true # endpoint: "http://10.0.0.5:8000" + # capacity: 8 + # max_queue: 32 + # queue_timeout_ms: 30000 + # request_timeout_ms: 300000 # cli: # enabled: true # profiles: diff --git a/packages/go/config/config.go b/packages/go/config/config.go index 0980233..36eb5ae 100644 --- a/packages/go/config/config.go +++ b/packages/go/config/config.go @@ -196,28 +196,44 @@ type AdaptersConf struct { // OllamaInstanceConf is one named Ollama engine instance within a node. type OllamaInstanceConf struct { - Name string `mapstructure:"name" yaml:"name"` - Enabled bool `mapstructure:"enabled" yaml:"enabled"` - BaseURL string `mapstructure:"base_url" yaml:"base_url"` - ContextSize int `mapstructure:"context_size" yaml:"context_size"` + Name string `mapstructure:"name" yaml:"name"` + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + BaseURL string `mapstructure:"base_url" yaml:"base_url"` + ContextSize int `mapstructure:"context_size" yaml:"context_size"` + Capacity int `mapstructure:"capacity" yaml:"capacity"` + MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } // VllmInstanceConf is one named vLLM engine instance within a node. type VllmInstanceConf struct { - Name string `mapstructure:"name" yaml:"name"` - Enabled bool `mapstructure:"enabled" yaml:"enabled"` - Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` + Name string `mapstructure:"name" yaml:"name"` + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` + Capacity int `mapstructure:"capacity" yaml:"capacity"` + MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } type OllamaConf struct { - Enabled bool `mapstructure:"enabled" yaml:"enabled"` - BaseURL string `mapstructure:"base_url" yaml:"base_url"` - ContextSize int `mapstructure:"context_size" yaml:"context_size"` + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + BaseURL string `mapstructure:"base_url" yaml:"base_url"` + ContextSize int `mapstructure:"context_size" yaml:"context_size"` + Capacity int `mapstructure:"capacity" yaml:"capacity"` + MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } type VllmConf struct { - Enabled bool `mapstructure:"enabled" yaml:"enabled"` - Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` + Enabled bool `mapstructure:"enabled" yaml:"enabled"` + Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` + Capacity int `mapstructure:"capacity" yaml:"capacity"` + MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } type CLIConf struct { @@ -312,25 +328,33 @@ func normalizeAdapters(a *AdaptersConf) error { existing := findOllamaInstance(a.OllamaInstances, "ollama") if existing == nil { a.OllamaInstances = append([]OllamaInstanceConf{{ - Name: "ollama", - Enabled: a.Ollama.Enabled, - BaseURL: a.Ollama.BaseURL, - ContextSize: a.Ollama.ContextSize, + Name: "ollama", + Enabled: a.Ollama.Enabled, + BaseURL: a.Ollama.BaseURL, + ContextSize: a.Ollama.ContextSize, + Capacity: a.Ollama.Capacity, + MaxQueue: a.Ollama.MaxQueue, + QueueTimeoutMS: a.Ollama.QueueTimeoutMS, + RequestTimeoutMS: a.Ollama.RequestTimeoutMS, }}, a.OllamaInstances...) - } else if !existing.Enabled || existing.BaseURL != a.Ollama.BaseURL || existing.ContextSize != a.Ollama.ContextSize { - return fmt.Errorf("ollama: legacy field conflicts with explicit instance %q: enabled/base_url/context_size mismatch", "ollama") + } else if !sameOllamaInstance(*existing, a.Ollama) { + return fmt.Errorf("ollama: legacy field conflicts with explicit instance %q: enabled/base_url/context_size/capacity/max_queue/queue_timeout_ms/request_timeout_ms mismatch", "ollama") } } if a.Vllm.Enabled { existing := findVllmInstance(a.VllmInstances, "vllm") if existing == nil { a.VllmInstances = append([]VllmInstanceConf{{ - Name: "vllm", - Enabled: a.Vllm.Enabled, - Endpoint: a.Vllm.Endpoint, + Name: "vllm", + Enabled: a.Vllm.Enabled, + Endpoint: a.Vllm.Endpoint, + Capacity: a.Vllm.Capacity, + MaxQueue: a.Vllm.MaxQueue, + QueueTimeoutMS: a.Vllm.QueueTimeoutMS, + RequestTimeoutMS: a.Vllm.RequestTimeoutMS, }}, a.VllmInstances...) - } else if !existing.Enabled || existing.Endpoint != a.Vllm.Endpoint { - return fmt.Errorf("vllm: legacy field conflicts with explicit instance %q: enabled/endpoint mismatch", "vllm") + } else if !sameVllmInstance(*existing, a.Vllm) { + return fmt.Errorf("vllm: legacy field conflicts with explicit instance %q: enabled/endpoint/capacity/max_queue/queue_timeout_ms/request_timeout_ms mismatch", "vllm") } } @@ -340,9 +364,56 @@ func normalizeAdapters(a *AdaptersConf) error { if err := checkUniqueNames("vllm_instances", func(i int) string { return a.VllmInstances[i].Name }, len(a.VllmInstances)); err != nil { return err } + for i, inst := range a.OllamaInstances { + field := fmt.Sprintf("ollama_instances[%d]", i) + if err := validateProviderQueueConfig(field, inst.Capacity, inst.MaxQueue, inst.QueueTimeoutMS, inst.RequestTimeoutMS); err != nil { + return err + } + } + for i, inst := range a.VllmInstances { + field := fmt.Sprintf("vllm_instances[%d]", i) + if err := validateProviderQueueConfig(field, inst.Capacity, inst.MaxQueue, inst.QueueTimeoutMS, inst.RequestTimeoutMS); err != nil { + return err + } + } return nil } +func validateProviderQueueConfig(field string, capacity, maxQueue, queueTimeoutMS, requestTimeoutMS int) error { + if capacity < 0 { + return fmt.Errorf("%s.capacity must be non-negative", field) + } + if maxQueue < 0 { + return fmt.Errorf("%s.max_queue must be non-negative", field) + } + if queueTimeoutMS < 0 { + return fmt.Errorf("%s.queue_timeout_ms must be non-negative", field) + } + if requestTimeoutMS < 0 { + return fmt.Errorf("%s.request_timeout_ms must be non-negative", field) + } + return nil +} + +func sameOllamaInstance(inst OllamaInstanceConf, legacy OllamaConf) bool { + return inst.Enabled == legacy.Enabled && + inst.BaseURL == legacy.BaseURL && + inst.ContextSize == legacy.ContextSize && + inst.Capacity == legacy.Capacity && + inst.MaxQueue == legacy.MaxQueue && + inst.QueueTimeoutMS == legacy.QueueTimeoutMS && + inst.RequestTimeoutMS == legacy.RequestTimeoutMS +} + +func sameVllmInstance(inst VllmInstanceConf, legacy VllmConf) bool { + return inst.Enabled == legacy.Enabled && + inst.Endpoint == legacy.Endpoint && + inst.Capacity == legacy.Capacity && + inst.MaxQueue == legacy.MaxQueue && + inst.QueueTimeoutMS == legacy.QueueTimeoutMS && + inst.RequestTimeoutMS == legacy.RequestTimeoutMS +} + func findOllamaInstance(instances []OllamaInstanceConf, name string) *OllamaInstanceConf { for i := range instances { if instances[i].Name == name { diff --git a/packages/go/config/config_test.go b/packages/go/config/config_test.go index 97a35cd..32787f0 100644 --- a/packages/go/config/config_test.go +++ b/packages/go/config/config_test.go @@ -162,6 +162,10 @@ nodes: enabled: true base_url: "http://192.168.0.91:11434" context_size: 262144 + capacity: 3 + max_queue: 8 + queue_timeout_ms: 1500 + request_timeout_ms: 30000 ` if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { t.Fatalf("write yaml: %v", err) @@ -178,6 +182,9 @@ nodes: if ollama.BaseURL != "http://192.168.0.91:11434" || ollama.ContextSize != 262144 { t.Fatalf("unexpected ollama config: %+v", ollama) } + if ollama.Capacity != 3 || ollama.MaxQueue != 8 || ollama.QueueTimeoutMS != 1500 || ollama.RequestTimeoutMS != 30000 { + t.Fatalf("unexpected ollama queue config: %+v", ollama) + } } func TestLoadEdge_ConsoleSessionDefaults(t *testing.T) { @@ -935,10 +942,18 @@ nodes: enabled: true base_url: "http://127.0.0.1:11434" context_size: 131072 + capacity: 2 + max_queue: 4 + queue_timeout_ms: 1000 + request_timeout_ms: 20000 - name: "dgx" enabled: true base_url: "http://192.168.0.91:11434" context_size: 262144 + capacity: 6 + max_queue: 12 + queue_timeout_ms: 2000 + request_timeout_ms: 60000 ` if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { t.Fatalf("write yaml: %v", err) @@ -957,9 +972,15 @@ nodes: if insts[0].Name != "local" || insts[0].BaseURL != "http://127.0.0.1:11434" || insts[0].ContextSize != 131072 { t.Errorf("unexpected instance[0]: %+v", insts[0]) } + if insts[0].Capacity != 2 || insts[0].MaxQueue != 4 || insts[0].QueueTimeoutMS != 1000 || insts[0].RequestTimeoutMS != 20000 { + t.Errorf("unexpected instance[0] queue config: %+v", insts[0]) + } if insts[1].Name != "dgx" || insts[1].BaseURL != "http://192.168.0.91:11434" || insts[1].ContextSize != 262144 { t.Errorf("unexpected instance[1]: %+v", insts[1]) } + if insts[1].Capacity != 6 || insts[1].MaxQueue != 12 || insts[1].QueueTimeoutMS != 2000 || insts[1].RequestTimeoutMS != 60000 { + t.Errorf("unexpected instance[1] queue config: %+v", insts[1]) + } } func TestLoadEdge_MultiVllmInstances(t *testing.T) { @@ -975,9 +996,17 @@ nodes: - name: "a100" enabled: true endpoint: "http://10.0.0.5:8000" + capacity: 4 + max_queue: 10 + queue_timeout_ms: 1500 + request_timeout_ms: 45000 - name: "h100" enabled: true endpoint: "http://10.0.0.6:8000" + capacity: 8 + max_queue: 16 + queue_timeout_ms: 2500 + request_timeout_ms: 90000 ` if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { t.Fatalf("write yaml: %v", err) @@ -996,9 +1025,15 @@ nodes: if insts[0].Name != "a100" || insts[0].Endpoint != "http://10.0.0.5:8000" { t.Errorf("unexpected instance[0]: %+v", insts[0]) } + if insts[0].Capacity != 4 || insts[0].MaxQueue != 10 || insts[0].QueueTimeoutMS != 1500 || insts[0].RequestTimeoutMS != 45000 { + t.Errorf("unexpected instance[0] queue config: %+v", insts[0]) + } if insts[1].Name != "h100" || insts[1].Endpoint != "http://10.0.0.6:8000" { t.Errorf("unexpected instance[1]: %+v", insts[1]) } + if insts[1].Capacity != 8 || insts[1].MaxQueue != 16 || insts[1].QueueTimeoutMS != 2500 || insts[1].RequestTimeoutMS != 90000 { + t.Errorf("unexpected instance[1] queue config: %+v", insts[1]) + } } func TestLoadEdge_LegacyOllamaPromotedToInstances(t *testing.T) { @@ -1014,6 +1049,10 @@ nodes: enabled: true base_url: "http://192.168.0.91:11434" context_size: 262144 + capacity: 5 + max_queue: 9 + queue_timeout_ms: 1700 + request_timeout_ms: 55000 ` if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { t.Fatalf("write yaml: %v", err) @@ -1032,6 +1071,9 @@ nodes: if insts[0].BaseURL != "http://192.168.0.91:11434" { t.Errorf("expected base_url %q, got %q", "http://192.168.0.91:11434", insts[0].BaseURL) } + if insts[0].Capacity != 5 || insts[0].MaxQueue != 9 || insts[0].QueueTimeoutMS != 1700 || insts[0].RequestTimeoutMS != 55000 { + t.Errorf("unexpected promoted queue config: %+v", insts[0]) + } } func TestLoadEdge_DuplicateOllamaInstanceNameRejected(t *testing.T) { @@ -1060,6 +1102,33 @@ nodes: } } +func TestLoadEdge_ProviderQueueConfigRejectsNegative(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "edge.yaml") + yaml := ` +server: + listen: "0.0.0.0:9090" +nodes: + - alias: "bad-queue" + adapters: + ollama_instances: + - name: "local" + enabled: true + base_url: "http://127.0.0.1:11434" + capacity: -1 +` + if err := os.WriteFile(f, []byte(yaml), 0o600); err != nil { + t.Fatalf("write yaml: %v", err) + } + _, err := config.LoadEdge(f) + if err == nil { + t.Fatal("expected negative provider queue config error") + } + if !strings.Contains(err.Error(), "capacity") || !strings.Contains(err.Error(), "non-negative") { + t.Fatalf("expected error to mention capacity non-negative, got %v", err) + } +} + func TestLoadEdge_OpenAIRouteCatalog(t *testing.T) { dir := t.TempDir() f := filepath.Join(dir, "edge.yaml") diff --git a/proto/gen/iop/runtime.pb.go b/proto/gen/iop/runtime.pb.go index 7a01f98..9906bf7 100644 --- a/proto/gen/iop/runtime.pb.go +++ b/proto/gen/iop/runtime.pb.go @@ -1718,11 +1718,15 @@ func (x *CLICompletionMarker) GetRegex() string { } type OllamaAdapterConfig struct { - state protoimpl.MessageState `protogen:"open.v1"` - BaseUrl string `protobuf:"bytes,1,opt,name=base_url,json=baseUrl,proto3" json:"base_url,omitempty"` - ContextSize int32 `protobuf:"varint,2,opt,name=context_size,json=contextSize,proto3" json:"context_size,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + BaseUrl string `protobuf:"bytes,1,opt,name=base_url,json=baseUrl,proto3" json:"base_url,omitempty"` + ContextSize int32 `protobuf:"varint,2,opt,name=context_size,json=contextSize,proto3" json:"context_size,omitempty"` + Capacity int32 `protobuf:"varint,3,opt,name=capacity,proto3" json:"capacity,omitempty"` + MaxQueue int32 `protobuf:"varint,4,opt,name=max_queue,json=maxQueue,proto3" json:"max_queue,omitempty"` + QueueTimeoutMs int32 `protobuf:"varint,5,opt,name=queue_timeout_ms,json=queueTimeoutMs,proto3" json:"queue_timeout_ms,omitempty"` + RequestTimeoutMs int32 `protobuf:"varint,6,opt,name=request_timeout_ms,json=requestTimeoutMs,proto3" json:"request_timeout_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *OllamaAdapterConfig) Reset() { @@ -1769,11 +1773,43 @@ func (x *OllamaAdapterConfig) GetContextSize() int32 { return 0 } +func (x *OllamaAdapterConfig) GetCapacity() int32 { + if x != nil { + return x.Capacity + } + return 0 +} + +func (x *OllamaAdapterConfig) GetMaxQueue() int32 { + if x != nil { + return x.MaxQueue + } + return 0 +} + +func (x *OllamaAdapterConfig) GetQueueTimeoutMs() int32 { + if x != nil { + return x.QueueTimeoutMs + } + return 0 +} + +func (x *OllamaAdapterConfig) GetRequestTimeoutMs() int32 { + if x != nil { + return x.RequestTimeoutMs + } + return 0 +} + type VllmAdapterConfig struct { - state protoimpl.MessageState `protogen:"open.v1"` - Endpoint string `protobuf:"bytes,1,opt,name=endpoint,proto3" json:"endpoint,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Endpoint string `protobuf:"bytes,1,opt,name=endpoint,proto3" json:"endpoint,omitempty"` + Capacity int32 `protobuf:"varint,2,opt,name=capacity,proto3" json:"capacity,omitempty"` + MaxQueue int32 `protobuf:"varint,3,opt,name=max_queue,json=maxQueue,proto3" json:"max_queue,omitempty"` + QueueTimeoutMs int32 `protobuf:"varint,4,opt,name=queue_timeout_ms,json=queueTimeoutMs,proto3" json:"queue_timeout_ms,omitempty"` + RequestTimeoutMs int32 `protobuf:"varint,5,opt,name=request_timeout_ms,json=requestTimeoutMs,proto3" json:"request_timeout_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *VllmAdapterConfig) Reset() { @@ -1813,6 +1849,34 @@ func (x *VllmAdapterConfig) GetEndpoint() string { return "" } +func (x *VllmAdapterConfig) GetCapacity() int32 { + if x != nil { + return x.Capacity + } + return 0 +} + +func (x *VllmAdapterConfig) GetMaxQueue() int32 { + if x != nil { + return x.MaxQueue + } + return 0 +} + +func (x *VllmAdapterConfig) GetQueueTimeoutMs() int32 { + if x != nil { + return x.QueueTimeoutMs + } + return 0 +} + +func (x *VllmAdapterConfig) GetRequestTimeoutMs() int32 { + if x != nil { + return x.RequestTimeoutMs + } + return 0 +} + // NodeRuntimeConfig is the runtime tuning pushed to the node. type NodeRuntimeConfig struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -2034,12 +2098,20 @@ const file_proto_iop_runtime_proto_rawDesc = "" + "resumeArgs\"?\n" + "\x13CLICompletionMarker\x12\x12\n" + "\x04line\x18\x01 \x01(\tR\x04line\x12\x14\n" + - "\x05regex\x18\x02 \x01(\tR\x05regex\"S\n" + + "\x05regex\x18\x02 \x01(\tR\x05regex\"\xe4\x01\n" + "\x13OllamaAdapterConfig\x12\x19\n" + "\bbase_url\x18\x01 \x01(\tR\abaseUrl\x12!\n" + - "\fcontext_size\x18\x02 \x01(\x05R\vcontextSize\"/\n" + + "\fcontext_size\x18\x02 \x01(\x05R\vcontextSize\x12\x1a\n" + + "\bcapacity\x18\x03 \x01(\x05R\bcapacity\x12\x1b\n" + + "\tmax_queue\x18\x04 \x01(\x05R\bmaxQueue\x12(\n" + + "\x10queue_timeout_ms\x18\x05 \x01(\x05R\x0equeueTimeoutMs\x12,\n" + + "\x12request_timeout_ms\x18\x06 \x01(\x05R\x10requestTimeoutMs\"\xc0\x01\n" + "\x11VllmAdapterConfig\x12\x1a\n" + - "\bendpoint\x18\x01 \x01(\tR\bendpoint\"\\\n" + + "\bendpoint\x18\x01 \x01(\tR\bendpoint\x12\x1a\n" + + "\bcapacity\x18\x02 \x01(\x05R\bcapacity\x12\x1b\n" + + "\tmax_queue\x18\x03 \x01(\x05R\bmaxQueue\x12(\n" + + "\x10queue_timeout_ms\x18\x04 \x01(\x05R\x0equeueTimeoutMs\x12,\n" + + "\x12request_timeout_ms\x18\x05 \x01(\x05R\x10requestTimeoutMs\"\\\n" + "\x11NodeRuntimeConfig\x12 \n" + "\vconcurrency\x18\x01 \x01(\x05R\vconcurrency\x12%\n" + "\x0eworkspace_root\x18\x02 \x01(\tR\rworkspaceRoot*\x81\x01\n" + diff --git a/proto/iop/runtime.proto b/proto/iop/runtime.proto index 3c1b659..08f18af 100644 --- a/proto/iop/runtime.proto +++ b/proto/iop/runtime.proto @@ -203,12 +203,20 @@ message CLICompletionMarker { } message OllamaAdapterConfig { - string base_url = 1; - int32 context_size = 2; + string base_url = 1; + int32 context_size = 2; + int32 capacity = 3; + int32 max_queue = 4; + int32 queue_timeout_ms = 5; + int32 request_timeout_ms = 6; } message VllmAdapterConfig { - string endpoint = 1; + string endpoint = 1; + int32 capacity = 2; + int32 max_queue = 3; + int32 queue_timeout_ms = 4; + int32 request_timeout_ms = 5; } // NodeRuntimeConfig is the runtime tuning pushed to the node.