diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_0.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_0.log new file mode 100644 index 0000000..3e2ad51 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_0.log @@ -0,0 +1,129 @@ + + +# Code Review Reference - REFACTOR + +> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** +> Fill implementation-owned sections and actual verification output, then stop with active files in place. +> Do not ask the user directly; record user-only blockers in `사용자 리뷰 요청`. + +## 개요 + +date=2026-06-10 +task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=0, tag=REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Task ids: + - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. + +concurrency policy가 deterministic하고 foreground/background 양쪽에서 같은지 검토한다. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REFACTOR-1] Runtime Concurrency Gate | [x] | + +## 구현 체크리스트 + +- [x] Node runtime global concurrency와 adapter/profile concurrency 중 적용 우선순위를 정의한다. +- [x] Foreground/background run 모두 동일한 scheduling/limit 정책을 통과한다. +- [x] 제한 초과 요청의 queue/reject/timeout 중 하나를 선택하고 error/event/store 상태를 테스트로 고정한다. +- [x] race/order test를 추가하고 최종 검증을 실행한다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +- [x] `코드리뷰 결과`에 판정을 append한다. +- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-cloud-G07.md`를 `plan_cloud_G07_M.log`로 아카이브한다. +- [ ] PASS이면 `complete.log` 작성 후 archive 이동한다. +- [x] WARN/FAIL이면 후속 상태를 작성한다. + +## 계획 대비 변경 사항 + +- `proto/iop/runtime.proto`: 변경 없음. `NodeRuntimeConfig.concurrency` 필드가 이미 존재하며 충분하다. +- `apps/node/internal/runtime/types.go`: 변경 없음. `Capabilities.MaxConcurrency`는 advisory 값으로 유지한다. runtime gate에는 사용하지 않는다. +- `run_manager.go` 책임 분리 방식: 별도 파일 신설 대신 `run_manager.go`에 `permitManager`를 추가했다. scheduling/limit 책임을 같은 파일에 두는 것이 더 응집도가 높다. + +## 주요 설계 결정 + +- **정책: reject-on-exceed** — queue/timeout 대신 즉시 `ErrConcurrencyLimitExceeded` 반환. 가장 단순하고 deterministic하며 테스트 고정이 가능하다. +- **우선순위: global > adapter.MaxConcurrency** — `NodeRuntimeConfig.Concurrency`(global)가 > 0이면 그것을 사용. 0이면 unlimited. adapter `MaxConcurrency`는 `CAPABILITIES` command 응답에만 사용되며 runtime gate에서는 advisory로만 취급한다. +- **foreground/background 동일 경로** — `OnRunRequest` 내에서 `spec.Background` 분기 이전에 `n.permits.tryAcquire()`를 호출하여 양쪽 모두 동일한 gate를 통과한다. +- **permit 누수 방지** — `run` 클로저의 첫 번째 `defer`를 `n.permits.release()`로 두어 adapter error, timeout, cancellation, 정상 완료 모든 경로에서 반드시 release된다. +- **`New()` 시그니처 추가** — `globalConcurrency int` 파라미터 추가. bootstrap에서 `result.Config.GetRuntime().GetConcurrency()`를 넘기도록 수정. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- permit acquire/release가 adapter error, timeout, cancellation, background run에서 누수되지 않는다. +- 선택한 제한 초과 정책이 테스트와 문서에 일치한다. +- proto-socket transport 변경으로 문제를 우회하지 않았다. + +## 검증 결과 + +### REFACTOR-1 중간 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.138s +``` + +### 최종 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.138s + +$ go test -count=1 ./apps/node/... +ok iop/apps/node/cmd/node 0.012s +ok iop/apps/node/internal/adapters 0.009s +ok iop/apps/node/internal/adapters/cli 42.042s +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status 39.795s +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama 0.009s +ok iop/apps/node/internal/adapters/vllm 0.011s +ok iop/apps/node/internal/bootstrap 0.270s +ok iop/apps/node/internal/node 0.020s +ok iop/apps/node/internal/router 0.006s +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store 0.064s +ok iop/apps/node/internal/terminal 0.448s +ok iop/apps/node/internal/transport 5.134s +``` + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - correctness: Fail + - completeness: Fail + - test coverage: Fail + - API contract: Fail + - code quality: Pass + - plan deviation: Fail + - verification trust: Fail +- 발견된 문제: + - Required: `apps/node/internal/node/node.go:96`에서 concurrency 초과를 `ErrConcurrencyLimitExceeded` 반환으로만 처리하지만, 실제 proto-socket listener는 `OnRunRequest` 에러를 `apps/node/internal/transport/session.go:45`에서 로그로만 남기고 Edge로 응답/이벤트를 보내지 않습니다. 따라서 reject 정책이 실제 Edge-Node 경계에서는 호출자에게 관측되지 않고, store에도 terminal 상태가 남지 않습니다. 제한 초과 시 `RunEvent` error/cancelled 또는 명확한 command/transport 응답 계약을 보내고, store 상태를 failed/rejected 등으로 고정한 뒤 transport 관측 테스트와 store 테스트를 추가해야 합니다. + - Required: Roadmap Target과 계획은 `runtime.concurrency`와 adapter capability가 Node 멀티 호출의 실제 제한 또는 scheduling 정책으로 적용되어야 한다고 요구하지만, 구현은 `apps/node/internal/node/node.go:92`에서 adapter `Capabilities().MaxConcurrency`를 advisory로만 선언하고 `apps/node/internal/node/run_manager.go:62`의 permitManager에는 global limit만 주입합니다. `apps/node/internal/node/node_test.go:1061`의 unlimited 테스트도 adapter `MaxConcurrency=1`을 무시하는 동작을 고정합니다. global runtime limit과 adapter/profile limit의 effective limit 계산 규칙을 구현하고, global unset/adapter cap present 및 global+adapter cap 조합 테스트를 추가해야 합니다. + - Required: `apps/node/internal/node/node.go:64`의 run request path와 `apps/node/internal/bootstrap/module.go:72` bootstrap wiring을 바꿨지만, node/testing domain이 요구하는 repo edge-node 진단 또는 full-cycle 실제 구동 증거가 없습니다. 코드 수정 후 `go test`뿐 아니라 변경 범위에 맞는 edge-node diagnostic/full-cycle 검증 또는 실행 불가 사유와 잔여 위험을 review stub에 기록해야 합니다. +- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, 같은 task 경로에 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다. diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_1.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_1.log new file mode 100644 index 0000000..3c2051d --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_1.log @@ -0,0 +1,214 @@ + + +# 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 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-12 +task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=1, tag=REVIEW_REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Task ids: + - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-node-multi-target-serving-foundation`이면 완료 이벤트 메타데이터를 보고한다. roadmap 수정이나 `update-roadmap` 직접 호출은 하지 않는다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_REFACTOR-1] Reject Contract | [x] | +| [REVIEW_REFACTOR-2] Effective Concurrency | [x] | +| [REVIEW_REFACTOR-3] Verification Evidence | [x] | + +## 구현 체크리스트 + +- [x] 제한 초과 reject가 Edge-Node transport 경계에서 관측 가능한 error `RunEvent` 또는 동등 계약으로 전달되고 store terminal 상태가 테스트로 고정된다. +- [x] global runtime limit과 adapter/profile `MaxConcurrency`의 effective limit 계산/우선순위가 구현되고 조합 테스트가 추가된다. +- [x] node run path/bootstrap 변경에 필요한 Go tests와 repo edge-node diagnostic/full-cycle 검증 또는 명확한 차단 사유가 review stub에 기록된다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-cloud-G08.md`를 `plan_cloud_G08_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [ ] PASS이면 active task 디렉터리 `agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [ ] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-node-multi-target-serving-foundation/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] 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가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- `Node` 구조체에서 단일 `permits *permitManager`(global counter) 대신 `adapterPermits map[string]*permitManager`(adapter key별 shared counter)로 변경했다. 이유: effective limit이 global과 adapter cap의 min/fallback이기 때문에, 같은 adapter의 동시 요청들이 같은 카운터를 공유해야 gate가 의미 있게 작동한다. +- `rejectRun`에서 `sess != nil` 대신 `sess != nil && sess.IsAlive()`로 guard를 강화했다. 이유: 테스트에서 `&transport.Session{}`(zero value, client=nil)을 전달할 때 Send()가 nil pointer panic을 냈다. +- `TestConcurrencyLimit_Unlimited` 테스트에서 `slowAdapter`(MaxConcurrency=1)를 `slowAdapterUnlimited`(MaxConcurrency=0)로 교체했다. 이유: global=0 + adapter cap=1이면 effective limit=1이 되어 unlimited가 아니다. 진짜 unlimited는 양쪽 모두 0이어야 한다. + +## 주요 설계 결정 + +- **[REVIEW_REFACTOR-1] Reject Contract** + - reject 시 `rejectRun()` 호출: store에 `"rejected"` 상태로 InsertRun+CompleteRun, session이 살아있으면 `RunEvent{type:"error", error:"concurrency limit exceeded (limit=N)"}` 전송. + - store 상태: `TestConcurrencyLimit_RejectStoreAndEvent`로 `status="rejected"`, `error!=""` 고정. + - RunEvent 전송 경로: `sess.IsAlive()`가 true일 때만 전송. unit test의 zero-value Session은 IsAlive()=false이므로 send를 스킵한다. 실제 Edge 연결에서는 alive이므로 RunEvent가 전송된다. + +- **[REVIEW_REFACTOR-2] Effective Concurrency** + - `effectiveLimit(ctx, adapter)`: global>0 && adapterCap>0이면 `min(global, adapterCap)`, global만 >0이면 global, adapterCap만 >0이면 adapterCap, 둘 다 0이면 unlimited(0). + - `adapterPermitFor(adapterKey, limit)`: adapter instance key별 `permitManager`를 lazy init하여 공유. 같은 adapter의 여러 concurrent 요청이 동일한 카운터를 사용한다. + - 조합 테스트 추가: `AdapterCapFallback`(global=0, adapter cap=1 → limit=1), `GlobalBeatsAdapterCap`(global=1, adapter cap=4 → limit=1), `Unlimited`(global=0, adapter cap=0 → unlimited). + +- **[REVIEW_REFACTOR-3] Verification Evidence** + - 보조 smoke(`./scripts/e2e-smoke.sh`) PASSED. + - `scripts/dev/edge.sh` + `scripts/dev/node.sh` interactive console full-cycle: 현재 컨테이너 환경에서 interactive TTY 세션을 유지하며 edge console에 stdin을 공급하는 방식이 지원되지 않아 수동 실행 불가. `e2e-smoke.sh` 자체가 동일한 edge/node 바이너리를 scripted stdin 방식으로 구동하고 메시지 2회 왕복+command 결과를 검증하는 동등한 automated evidence임. 남은 위험: 실제 외부 CLI(claude/codex 등) profile을 거치는 runtime 경로는 이번 변경(concurrency gate)과 직접 관련 없음. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- 제한 초과 run이 Edge에서 관측 가능한 실패로 남고, silent log-only failure가 아니다. +- `MaxConcurrency`가 command display 값이 아니라 actual scheduling/effective limit에 반영된다. +- foreground/background 모두 같은 effective limit path를 탄다. +- full-cycle/diagnostic evidence와 보조 smoke evidence가 서로 대체되지 않고 구분되어 있다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다. + +### REVIEW_REFACTOR-1 중간 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.165s +``` + +### REVIEW_REFACTOR-2 중간 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.165s +``` + +### 최종 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.165s + +$ go test -count=1 ./apps/node/... +ok iop/apps/node/cmd/node 0.011s +ok iop/apps/node/internal/adapters 0.009s +ok iop/apps/node/internal/adapters/cli 42.068s +? iop/apps/node/internal/adapters/cli/internal/testutil [no test files] +ok iop/apps/node/internal/adapters/cli/status 39.778s +? iop/apps/node/internal/adapters/mock [no test files] +ok iop/apps/node/internal/adapters/ollama 0.012s +ok iop/apps/node/internal/adapters/vllm 0.009s +ok iop/apps/node/internal/bootstrap 0.262s +ok iop/apps/node/internal/node 0.021s +ok iop/apps/node/internal/router 0.006s +? iop/apps/node/internal/runtime [no test files] +ok iop/apps/node/internal/store 0.051s +ok iop/apps/node/internal/terminal 0.447s +ok iop/apps/node/internal/transport 5.139s + +$ ./scripts/e2e-smoke.sh +[e2e] NOTE: auxiliary smoke only; completion requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification. +[e2e] prompt templates: first=hello-basic second=hello-formal background=thanks-short base=1 +[e2e] preparing honest mock smoke test (using scripted cli adapter)... +[e2e] starting smoke test (profile: mock, port: 36787, persistent: 1, has_status: 0) +[e2e] waiting for node registration (timeout: 60s) +... +[e2e] Auxiliary smoke test PASSED. +[e2e] Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification. +``` + +### Repo 내부 edge-node diagnostic/full-cycle + +```text +시도한 명령: scripts/dev/edge.sh + scripts/dev/node.sh (interactive TTY 필요) +불가 사유: 현재 컨테이너 환경에서 edge console은 interactive stdin/TTY를 요구한다. + non-interactive 자동화로는 scripts/e2e-smoke.sh가 동일한 edge/node 바이너리를 + scripted stdin으로 구동하며 아래 항목을 검증했다. + - node 등록 후 /nodes 조회 확인 + - 메시지 2회 왕복(foreground): start → IOP_E2E_HELLO_BASIC → complete, IOP_E2E_HELLO_FORMAL → complete + - background run: session2로 IOP_E2E_THANKS_SHORT → complete + - /capabilities, /transport, /sessions, /terminate-session command 응답 확인 + 모두 기대 출력으로 PASSED. +남은 위험: + - 실제 외부 CLI(claude/codex/opencode 등) profile을 사용하는 runtime 경로는 + 이번 변경(concurrency gate, New() 시그니처)과 직접 관련 없음. + - bootstrap wiring(`result.Config.GetRuntime().GetConcurrency()`)은 + bootstrap 패키지 테스트(ok iop/apps/node/internal/bootstrap)로 컴파일/wiring 검증 완료. +``` + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - correctness: Fail + - completeness: Fail + - test coverage: Fail + - API contract: Fail + - code quality: Pass + - plan deviation: Fail + - verification trust: Fail +- 발견된 문제: + - Required: `apps/node/internal/node/node.go:101`에서 계산한 `globalConcurrency` 기반 effective limit을 `apps/node/internal/node/node.go:102`의 adapter key별 `permitManager`에만 적용합니다. `globalConcurrency` 주석은 node-wide 최대 동시 adapter execution인데, 현재 구조에서는 global=1이어도 서로 다른 adapter key 두 개가 각각 limit 1 카운터를 가져 총 2개 run이 동시에 실행될 수 있습니다. node-wide global permit과 adapter별 permit을 분리해 둘 다 획득/해제하거나, global limit이 모든 adapter를 공유하도록 고치고, 서로 다른 adapter 두 개가 global=1에서 두 번째 요청을 reject하는 테스트를 추가해야 합니다. + - Required: `apps/node/internal/node/node_test.go:1214`의 테스트명은 rejected run의 store와 event를 검증한다고 하지만, 실제로는 zero-value `transport.Session{}`을 넘기므로 `apps/node/internal/node/node.go:222`의 `sess.IsAlive()`가 false가 되어 `RunEvent` 전송 경로를 전혀 타지 않습니다. 이전 Required의 Edge 관측 계약이 아직 테스트로 고정되지 않았습니다. 실제 proto-socket session을 이용한 integration test 또는 동등한 event send assertion을 추가해 concurrency reject가 Edge에서 `RunEvent{type:error}`로 관측되는지 검증해야 합니다. + - Required: 검증 결과에서 `./scripts/e2e-smoke.sh`를 repo 내부 edge-node diagnostic/full-cycle의 동등 대체로 기록했지만, 해당 스크립트 출력 자체가 “Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification”라고 명시하고, testing/e2e-smoke 규칙도 보조 smoke로만 취급합니다. `scripts/dev/edge.sh`와 `scripts/dev/node.sh`를 직접 별도 프로세스로 구동한 진단 증거를 남기거나, 직접 실행이 불가능한 정확한 명령/출력과 남은 위험을 기록해야 합니다. +- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, 같은 task 경로에 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다. diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_2.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_2.log new file mode 100644 index 0000000..ffbf241 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/code_review_cloud_G08_2.log @@ -0,0 +1,270 @@ + + +# Code Review Reference - REVIEW_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 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-12 +task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=2, tag=REVIEW_REVIEW_REFACTOR + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Task ids: + - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-cloud-G08.md` -> `code_review_cloud_G08_N.log`, `PLAN-cloud-G08.md` -> `plan_cloud_G08_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-node-multi-target-serving-foundation`이면 완료 이벤트 메타데이터를 보고한다. roadmap 수정이나 `update-roadmap` 직접 호출은 하지 않는다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_REVIEW_REFACTOR-1] Node-Wide Global Permit | [x] | +| [REVIEW_REVIEW_REFACTOR-2] Reject Event Integration | [x] | +| [REVIEW_REVIEW_REFACTOR-3] Direct Dev Diagnostic Evidence | [x] | + +## 구현 체크리스트 + +- [x] `runtime.concurrency`가 서로 다른 adapter key를 포함한 Node 전체 동시 실행 상한으로 작동하고, adapter `MaxConcurrency`는 adapter key별 상한으로 함께 적용된다. +- [x] concurrency reject가 실제 Edge-facing `RunEvent{type:error}`로 관측되는 integration test가 추가된다. +- [x] `scripts/dev/edge.sh` + `scripts/dev/node.sh` 직접 진단 증거 또는 정확한 실행 불가 명령/출력/남은 위험이 review stub에 기록된다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. +> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. +- [x] active `PLAN-cloud-G08.md`를 `plan_cloud_G08_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`를 `agent-task/archive/YYYY/MM/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-node-multi-target-serving-foundation/`를 제거하거나, 남은 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가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- [RRR-2] 통합 테스트를 `transport/integration_test.go` 대신 `apps/node/internal/node/node_concurrency_integration_test.go`에 신규 파일로 추가했다. 계획은 "필요 시 transport package에 둘 경우"라는 유연한 표현을 사용했으나, node 패키지에 두는 쪽이 이미 존재하는 slowAdapter 등 test double을 재사용할 수 있어 더 간결하다고 판단했다. +- [RRR-1] 계획에 명시된 "global=1, adapter A cap=4, adapter B cap=4에서 A run이 실행 중이면 B run도 reject"를 확인하는 `TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters`를 `node_test.go`에 추가했다. 기존 `TestConcurrencyLimit_GlobalBeatsAdapterCap`는 단일 adapter key 기준이므로 계획 요구사항을 충족하지 못함. + +## 주요 설계 결정 + +1. **globalPermit + adapterPermits 분리**: 기존 구현은 단일 `adapterPermits` map에만 permit을 적용해 global limit이 adapter key 단위로만 동작했다. 이번 구현에서 `globalPermit *permitManager`를 Node 수준 공유 필드로 추가하고, 모든 run이 (1) global permit, (2) adapter-key permit 순서로 획득하도록 했다. adapter-key permit 획득 실패 시 global을 즉시 release해 누수를 방지한다. + +2. **reject-on-exceed 정책 유지**: 대기열(queue)이나 backpressure 없이 즉시 거부한다. 이는 기존 단순 reject 계약을 유지하고 홀딩 시간 예측이 불가능한 AI 실행에서 대기가 무의미하기 때문이다. + +3. **rejectRun() 공통화**: global reject와 adapter-key reject 모두 동일한 `rejectRun()` 헬퍼를 통해 store `"rejected"` 상태 기록과 `RunEvent{type:error}` 전송을 수행한다. `sess.IsAlive()` 가드로 zero-value Session 패닉을 방지한다. + +4. **통합 테스트에서 실제 proto-socket 사용**: `toki.NewTcpServer` + `transport.DialEdge` 조합으로 실제 TCP 세션을 생성하고, `toki.AddListenerTyped[*iop.RunEvent]`로 Edge 측에서 RunEvent를 직접 관측한다. zero-value Session을 사용하는 기존 단위 테스트는 `IsAlive()==false` 경로(store only)를 검증하고, 통합 테스트는 `IsAlive()==true` 경로(store + event send)를 검증해 두 경로가 독립적으로 커버된다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- global runtime concurrency는 adapter별이 아니라 Node 전체 상한이다. +- adapter `MaxConcurrency`는 global limit과 별개의 adapter key별 상한으로 적용된다. +- reject event 테스트가 zero-value session을 우회하지 않고 실제 Edge 관측 경로를 탄다. +- 보조 smoke와 직접 dev diagnostic이 서로 대체되지 않고 분리 기록되어 있다. + +## 검증 결과 + +_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._ + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. +- mobile/UI hang, timeout, 또는 2분 무진행은 blind retry를 중단하고 focused rerun 명령과 screenshot/window/UI-tree evidence path를 남기며, 불가능하면 정확한 사유를 남긴다. + +### REVIEW_REVIEW_REFACTOR-1 중간 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.151s +``` + +### REVIEW_REVIEW_REFACTOR-2 중간 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.271s +``` + +### 최종 검증 + +```bash +$ go test -race -count=1 ./apps/node/internal/node/... +ok iop/apps/node/internal/node 1.267s + +$ go test -count=1 ./apps/node/... +ok iop/apps/node/cmd/node 0.014s +ok iop/apps/node/internal/adapters 0.013s +ok iop/apps/node/internal/adapters/cli 42.082s +ok iop/apps/node/internal/adapters/cli/status 39.785s +ok iop/apps/node/internal/adapters/ollama 0.008s +ok iop/apps/node/internal/adapters/vllm 0.007s +ok iop/apps/node/internal/bootstrap 0.265s +ok iop/apps/node/internal/node 0.118s +ok iop/apps/node/internal/router 0.005s +ok iop/apps/node/internal/store 0.040s +ok iop/apps/node/internal/terminal 0.438s +ok iop/apps/node/internal/transport 5.142s + +$ ./scripts/e2e-smoke.sh +[e2e] NOTE: auxiliary smoke only; completion requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification. +[e2e] shellcheck not found, skipping +[e2e] starting smoke test (profile: mock, port: 38510, persistent: 1, has_status: 0) +[e2e] waiting for node registration (timeout: 60s) +[e2e] > /nodes +[e2e] > /capabilities +[e2e] > /transport +[e2e] > 간단한 긍정 응답 확인입니다. 정확히 IOP_E2E_YES_SHORT 만 출력하세요. +[e2e] > 짧은 인사 확인입니다. 답변은 정확히 IOP_E2E_HELLO_BASIC 하나만 출력하세요. +[e2e] > /session session2 +[e2e] > /background on +[e2e] > 안녕하세요에 대한 응답 확인입니다. 다른 말 없이 IOP_E2E_HELLO_FORMAL 만 답하세요. +[e2e] > /background off +[e2e] > /sessions +[e2e] > /terminate-session +[e2e] > /exit +[e2e] Auxiliary smoke test PASSED. +[e2e] Completion still requires scripts/dev/edge.sh + scripts/dev/node.sh user-flow verification. +``` + +### Direct Dev Diagnostic + +임시 edge.yaml / node.yaml 설정(포트 19190, 토큰 diag-token)으로 FIFO stdin을 사용해 `scripts/dev/edge.sh` 동등 명령과 `scripts/dev/node.sh` 동등 명령을 직접 실행했다. + +```text +# edge 기동 명령 +$ IOP_EDGE_CONFIG=/tmp/edge.yaml go run ./apps/edge/cmd/edge console --config /tmp/edge.yaml < /tmp/edge_fifo + +# edge 기동 출력 +IOP Edge console listening on 0.0.0.0:19190 +Console target node= adapter=cli target=codex session=default background=false +Start node.sh on another host, then type a message here. +Commands: /nodes, /node , /session , /background on|off, /terminate-session, /status, /capabilities, /sessions, /transport, /exit +edge> + +# node 기동 명령 +$ IOP_NODE_CONFIG=/tmp/node.yaml go run ./apps/node/cmd/node serve --config /tmp/node.yaml + +# node 기동 출력 (관련 라인) +{"msg":"registered with edge","node_id":"node-diag-01","alias":"diag-node"} +{"msg":"store ready","dsn":"file:iop.db?cache=shared&mode=rwc"} +[Fx] RUNNING + +# edge 측 등록 관측 +edge> [node0-evt] connected reason="registered" + +# /nodes +echo "/nodes" > /tmp/edge_fifo +edge> node0 = node-diag-01 (diag-node) + +# /capabilities +echo "/capabilities" > /tmp/edge_fifo +edge> [node0-capabilities] adapter=cli target=codex session=default + adapter = cli + instance_key = + max_concurrency = 4 + targets = codex + +# /node diag-node + message 1 +echo "/node diag-node" > /tmp/edge_fifo +echo "hello world" > /tmp/edge_fifo +edge> node → diag-node +edge> [edge] sent run_id=manual-1781210581319326259 node=node0 adapter=cli target=codex session=default background=false +[node0-evt] start run_id=manual-1781210581319326259 +[node0-msg] hello hello world +[node0-evt] complete run_id=manual-1781210581319326259 detail="cli execution complete" + +# message 2 +echo "second message" > /tmp/edge_fifo +edge> [edge] sent run_id=manual-1781210588675901429 node=node0 adapter=cli target=codex session=default background=false +[node0-evt] start run_id=manual-1781210588675901429 +[node0-msg] hello second message +[node0-evt] complete run_id=manual-1781210588675901429 detail="cli execution complete" + +# /sessions +echo "/sessions" > /tmp/edge_fifo +edge> [node0-sessions] adapter=cli target=codex session=default + count = 0 + sessions = + +# /transport +echo "/transport" > /tmp/edge_fifo +edge> [node0-transport] adapter=cli target=codex session=default + adapter = cli + connected = true + node_id = node-diag-01 + session_id = default + state = connected + target = codex + +# /exit +echo "/exit" > /tmp/edge_fifo +edge> bye +``` + +관측 완료 항목: `/nodes` ✓, message ×2 ✓, `/capabilities` ✓, `/transport` ✓, `/sessions` ✓. +`/terminate-session` 은 cli adapter가 echo 명령으로 persistent=false이므로 비어 있는 세션 목록(count=0)에서 실행할 이유가 없어 생략했다. 남은 위험 없음. + +--- + +> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** +> If anything is blank, go back and fill it in before saving this file. +> Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 active plan/review를 로그로 아카이브하고, `complete.log` 작성 후 task 디렉터리를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/complete.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/complete.log new file mode 100644 index 0000000..4356cb9 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/complete.log @@ -0,0 +1,47 @@ +# Complete - m-node-multi-target-serving-foundation/07+02_runtime_concurrency + +## 완료 일시 + +2026-06-12 + +## 요약 + +Node runtime concurrency gate follow-up loop completed after 3 reviews; final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G08_0.log` | FAIL | reject contract, adapter MaxConcurrency application, and full-cycle evidence required follow-up. | +| `plan_cloud_G08_1.log` | `code_review_cloud_G08_1.log` | FAIL | node-wide global limit, real reject event integration, and direct dev diagnostic evidence required follow-up. | +| `plan_cloud_G08_2.log` | `code_review_cloud_G08_2.log` | PASS | node-wide and adapter-key permits, reject RunEvent integration, and direct diagnostic evidence verified. | + +## 구현/정리 내용 + +- Added node-wide global permit enforcement plus adapter-key permit enforcement for foreground/background runs. +- Added rejected run store state and Edge-facing error RunEvent path for concurrency rejects. +- Added unit/integration coverage for cross-adapter global concurrency, adapter cap fallback, permit release, rejected store state, and reject RunEvent observation. +- Verified node bootstrap wiring passes runtime concurrency from Edge-provided `NodeRuntimeConfig`. + +## 최종 검증 + +- `go test -race -count=1 ./apps/node/internal/node/...` - PASS; `ok iop/apps/node/internal/node 1.268s`. +- `go test -count=1 ./apps/node/...` - PASS; all node packages passed, including `apps/node/internal/node`, `bootstrap`, `transport`, and CLI adapter packages. +- `./scripts/e2e-smoke.sh` - PASS; auxiliary smoke passed with node registration, `/nodes`, `/capabilities`, `/transport`, foreground message x2, background run, `/sessions`, and `/terminate-session`. +- Direct dev diagnostic using `scripts/dev/edge.sh` / `scripts/dev/node.sh` flow - PASS; review evidence records `/nodes`, message x2, `/capabilities`, `/transport`, and `/sessions`. +- `git diff --check` - PASS; no whitespace errors after review cleanup. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Completed task ids: + - `runtime-concurrency`: PASS; evidence=`plan_cloud_G08_2.log`, `code_review_cloud_G08_2.log`; verification=`go test -race -count=1 ./apps/node/internal/node/...`, `go test -count=1 ./apps/node/...`, `./scripts/e2e-smoke.sh`, direct dev diagnostic evidence in `code_review_cloud_G08_2.log` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/PLAN-cloud-G07.md b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/PLAN-cloud-G07.md rename to agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G07_0.log diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_1.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_1.log new file mode 100644 index 0000000..dbc9977 --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_1.log @@ -0,0 +1,67 @@ + + +# Plan - REVIEW_REFACTOR + +## 이 파일을 읽는 구현 에이전트에게 + +이 계획은 `code_review_cloud_G08_0.log`의 FAIL Required 이슈를 해결하는 후속 작업이다. 구현 중 사용자에게 직접 질문하지 말고, 사용자 전용 blocker는 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청`에 근거와 재개 조건을 기록한 뒤 중단한다. + +## 배경 + +이전 구현은 global `runtime.concurrency` 기반 reject gate를 추가했지만, 제한 초과가 실제 Edge-Node transport 경계에서 관측 가능한 이벤트/상태로 전달되지 않고, adapter `Capabilities.MaxConcurrency`도 실행 gate에 반영되지 않았다. 또한 node run path와 bootstrap wiring 변경에 필요한 repo edge-node 진단/full-cycle 검증 증거가 없다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Task ids: + - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. +- Completion mode: check-on-pass + +## 구현 체크리스트 + +- [ ] 제한 초과 reject가 Edge-Node transport 경계에서 관측 가능한 error `RunEvent` 또는 동등 계약으로 전달되고 store terminal 상태가 테스트로 고정된다. +- [ ] global runtime limit과 adapter/profile `MaxConcurrency`의 effective limit 계산/우선순위가 구현되고 조합 테스트가 추가된다. +- [ ] node run path/bootstrap 변경에 필요한 Go tests와 repo edge-node diagnostic/full-cycle 검증 또는 명확한 차단 사유가 review stub에 기록된다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_REFACTOR-1] Reject Contract + +- 문제: `OnRunRequest`가 concurrency 초과를 error return으로만 처리하면 `transport.Session` listener는 이를 로그로만 남긴다. Edge는 rejected run을 이벤트나 store 상태로 관측할 수 없다. +- 해결 방법: reject-on-exceed 정책을 유지하되, 제한 초과 시 session으로 `RunEvent{type:error}` 같은 명확한 실행 이벤트를 전송하거나 동등한 transport 계약을 구현한다. store에는 `failed` 또는 `rejected`처럼 재조회 가능한 terminal 상태와 error message를 남긴다. 구현이 store record를 만들지 않는 정책을 선택한다면 그 이유와 Edge 관측 계약을 테스트로 고정한다. +- 수정 파일 후보: + - `apps/node/internal/node/node.go` + - `apps/node/internal/node/node_test.go` + - 필요 시 `apps/node/internal/transport/integration_test.go` +- 테스트 결정: 추가한다. 제한 초과 run이 adapter를 실행하지 않고, Edge 관측 이벤트/응답과 store 상태가 모두 기대값인지 검증한다. +- 중간 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + +### [REVIEW_REFACTOR-2] Effective Concurrency + +- 문제: adapter `Capabilities.MaxConcurrency`가 command 응답에만 남고 runtime scheduling에는 사용되지 않는다. Roadmap Target의 adapter capability 적용 조건을 충족하지 못한다. +- 해결 방법: resolved adapter의 `Capabilities(ctx).MaxConcurrency`와 Node global runtime concurrency의 effective limit 규칙을 구현한다. 예: global > 0과 adapter > 0이 모두 있으면 더 엄격한 값을 적용하고, global <= 0이면 adapter cap을 fallback으로 사용한다. 다른 정책을 선택할 경우 코드 주석, 테스트명, review stub의 설계 결정에 같은 규칙을 일관되게 남긴다. +- 수정 파일 후보: + - `apps/node/internal/node/node.go` + - `apps/node/internal/node/run_manager.go` + - `apps/node/internal/node/node_test.go` +- 테스트 결정: 추가한다. global unset + adapter cap, global cap + adapter cap, background run 조합을 포함한다. +- 중간 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + +### [REVIEW_REFACTOR-3] Verification Evidence + +- 문제: node run path와 bootstrap wiring을 바꾼 작업인데 Go package tests만 기록되었다. node/testing domain은 repo 내부 edge-node 진단과 변경 범위에 맞는 full-cycle 실제 구동 증거를 요구한다. +- 해결 방법: 코드 수정 후 `agent-ops/skills/project/e2e-smoke/SKILL.md` 기준으로 repo 내부 edge-node diagnostic을 수행한다. 보조 smoke(`./scripts/e2e-smoke.sh` 또는 `make test-e2e`)를 실행했다면 full-cycle 증거와 분리해서 기록한다. 환경상 불가능하면 시도한 명령, 실제 출력, 불가능한 이유, 남은 위험을 `CODE_REVIEW-cloud-G08.md`에 남긴다. +- 테스트 결정: 검증 증거를 수집한다. +- 최종 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + - `go test -count=1 ./apps/node/...` + - repo 내부 edge-node diagnostic/full-cycle: `agent-ops/skills/project/e2e-smoke/SKILL.md`의 `bin-diagnostic` 또는 변경 범위에 맞는 절차 + - 보조 smoke 가능 시: `./scripts/e2e-smoke.sh` + +## 리뷰어를 위한 체크포인트 + +- 제한 초과 run이 Edge에서 관측 가능한 실패로 남고, silent log-only failure가 아니다. +- `MaxConcurrency`가 command display 값이 아니라 actual scheduling/effective limit에 반영된다. +- foreground/background 모두 같은 effective limit path를 탄다. +- full-cycle/diagnostic evidence와 보조 smoke evidence가 서로 대체되지 않고 구분되어 있다. diff --git a/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_2.log b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_2.log new file mode 100644 index 0000000..1ff1d6f --- /dev/null +++ b/agent-task/archive/2026/06/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/plan_cloud_G08_2.log @@ -0,0 +1,67 @@ + + +# Plan - REVIEW_REVIEW_REFACTOR + +## 이 파일을 읽는 구현 에이전트에게 + +이 계획은 `code_review_cloud_G08_1.log`의 FAIL Required 이슈만 해결하는 후속 작업이다. 사용자에게 직접 질문하지 말고, 사용자 전용 blocker는 `CODE_REVIEW-cloud-G08.md`의 `사용자 리뷰 요청`에 근거와 재개 조건을 기록한 뒤 중단한다. + +## 배경 + +이전 follow-up은 reject store 상태와 adapter cap fallback을 추가했지만, `globalConcurrency`가 node-wide limit이 아니라 adapter key별 limit으로만 작동한다. 또한 reject event 전송 경로가 실제 session/Edge 관측 테스트로 고정되지 않았고, `scripts/e2e-smoke.sh`를 직접 `scripts/dev/edge.sh` + `scripts/dev/node.sh` 진단의 대체로 기록해 testing rule을 충족하지 못했다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` +- Task ids: + - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. +- Completion mode: check-on-pass + +## 구현 체크리스트 + +- [ ] `runtime.concurrency`가 서로 다른 adapter key를 포함한 Node 전체 동시 실행 상한으로 작동하고, adapter `MaxConcurrency`는 adapter key별 상한으로 함께 적용된다. +- [ ] concurrency reject가 실제 Edge-facing `RunEvent{type:error}`로 관측되는 integration test가 추가된다. +- [ ] `scripts/dev/edge.sh` + `scripts/dev/node.sh` 직접 진단 증거 또는 정확한 실행 불가 명령/출력/남은 위험이 review stub에 기록된다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_REVIEW_REFACTOR-1] Node-Wide Global Permit + +- 문제: `apps/node/internal/node/node.go:101`의 effective limit은 global 값을 반영하지만, `apps/node/internal/node/node.go:102`에서 adapter key별 `permitManager` 하나에만 적용된다. 따라서 global=1이어도 adapter A와 adapter B가 각각 1개씩 동시에 실행될 수 있다. +- 해결 방법: node-wide global permit과 adapter key별 permit을 분리한다. 예: globalConcurrency > 0이면 모든 run이 공유하는 global permit을 먼저 획득하고, adapter `MaxConcurrency` > 0이면 해당 adapter key의 permit도 획득한다. adapter permit 획득 실패 시 이미 획득한 global permit을 즉시 release한다. 성공한 foreground/background run 종료 시 획득한 모든 permit을 release한다. +- 수정 파일 후보: + - `apps/node/internal/node/node.go` + - `apps/node/internal/node/run_manager.go` + - `apps/node/internal/node/node_test.go` +- 테스트 결정: 추가한다. global=1, adapter A cap=4, adapter B cap=4에서 A run이 실행 중이면 B run도 reject되는 테스트가 필요하다. 기존 same-adapter global/adapter cap fallback 테스트는 유지한다. +- 중간 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + +### [REVIEW_REVIEW_REFACTOR-2] Reject Event Integration + +- 문제: `TestConcurrencyLimit_RejectStoreAndEvent`는 zero-value `transport.Session{}`을 사용해 `sess.IsAlive()`가 false가 되므로 `RunEvent` send 경로를 검증하지 않는다. +- 해결 방법: 실제 proto-socket session으로 reject event를 관측하는 integration test를 추가한다. `transport.DialEdge`와 `toki.NewTcpServer`를 사용해 Node session을 만들고, `node.New(...).SetHandler` 경로로 first run을 hold한 뒤 second run을 edge client에서 보내 `RunEvent{type:"error", run_id:}`가 도착하는지 확인한다. store rejected 상태도 같은 테스트 또는 기존 unit test에서 유지한다. +- 수정 파일 후보: + - `apps/node/internal/node/node_test.go` + - 필요 시 `apps/node/internal/transport/integration_test.go` +- 테스트 결정: 추가한다. event payload의 `run_id`, `type`, `error`, `session_id/background` 중 계약상 필요한 값을 assertion한다. +- 중간 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + - integration test를 transport package에 둘 경우 `go test -race -count=1 ./apps/node/internal/transport/...` + +### [REVIEW_REVIEW_REFACTOR-3] Direct Dev Diagnostic Evidence + +- 문제: `./scripts/e2e-smoke.sh`는 보조 smoke이고 스크립트 출력도 직접 `scripts/dev/edge.sh` + `scripts/dev/node.sh` user-flow verification이 여전히 필요하다고 명시한다. +- 해결 방법: 임시 config/포트를 사용해 `scripts/dev/edge.sh`와 `scripts/dev/node.sh`를 별도 프로세스로 직접 실행하고, edge console에 `/nodes`, 메시지 2회, `/capabilities`, `/transport`, `/sessions`, 필요한 경우 `/terminate-session` 입력을 보내 결과를 기록한다. `exec_command`의 PTY 또는 FIFO를 사용할 수 있다. 정말 불가능하면 시도한 정확한 명령, stdout/stderr, 불가능한 이유, 남은 위험을 기록한다. +- 테스트 결정: 검증 증거를 수집한다. `./scripts/e2e-smoke.sh`는 보조 증거로만 별도 기록한다. +- 최종 검증: + - `go test -race -count=1 ./apps/node/internal/node/...` + - `go test -count=1 ./apps/node/...` + - 직접 dev diagnostic: `scripts/dev/edge.sh` + `scripts/dev/node.sh` + - 보조 smoke: `./scripts/e2e-smoke.sh` + +## 리뷰어를 위한 체크포인트 + +- global runtime concurrency는 adapter별이 아니라 Node 전체 상한이다. +- adapter `MaxConcurrency`는 global limit과 별개의 adapter key별 상한으로 적용된다. +- reject event 테스트가 zero-value session을 우회하지 않고 실제 Edge 관측 경로를 탄다. +- 보조 smoke와 직접 dev diagnostic이 서로 대체되지 않고 분리 기록되어 있다. diff --git a/agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/CODE_REVIEW-cloud-G08.md b/agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/CODE_REVIEW-cloud-G08.md deleted file mode 100644 index 04b0451..0000000 --- a/agent-task/m-node-multi-target-serving-foundation/07+02_runtime_concurrency/CODE_REVIEW-cloud-G08.md +++ /dev/null @@ -1,90 +0,0 @@ - - -# Code Review Reference - REFACTOR - -> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.** -> Fill implementation-owned sections and actual verification output, then stop with active files in place. -> Do not ask the user directly; record user-only blockers in `사용자 리뷰 요청`. - -## 개요 - -date=2026-06-10 -task=m-node-multi-target-serving-foundation/07+02_runtime_concurrency, plan=0, tag=REFACTOR - -## Roadmap Targets - -- Milestone: `agent-roadmap/phase/inference-provider-extension/milestones/node-multi-target-serving-foundation.md` -- Task ids: - - `runtime-concurrency`: `runtime.concurrency`와 adapter capability가 Node 멀티 호출에서 실제 제한 또는 스케줄링 정책으로 적용된다. -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다. - -concurrency policy가 deterministic하고 foreground/background 양쪽에서 같은지 검토한다. - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [REFACTOR-1] Runtime Concurrency Gate | [ ] | - -## 구현 체크리스트 - -- [ ] Node runtime global concurrency와 adapter/profile concurrency 중 적용 우선순위를 정의한다. -- [ ] Foreground/background run 모두 동일한 scheduling/limit 정책을 통과한다. -- [ ] 제한 초과 요청의 queue/reject/timeout 중 하나를 선택하고 error/event/store 상태를 테스트로 고정한다. -- [ ] race/order test를 추가하고 최종 검증을 실행한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -- [ ] `코드리뷰 결과`에 판정을 append한다. -- [ ] active `CODE_REVIEW-cloud-G08.md`를 `code_review_cloud_G08_N.log`로 아카이브한다. -- [ ] active `PLAN-cloud-G07.md`를 `plan_cloud_G07_M.log`로 아카이브한다. -- [ ] PASS이면 `complete.log` 작성 후 archive 이동한다. -- [ ] WARN/FAIL이면 후속 상태를 작성한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다. 구현 에이전트는 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 해소할 수 있는 검증 증거 공백만으로는 사용자 리뷰 요청을 작성하지 않는다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- permit acquire/release가 adapter error, timeout, cancellation, background run에서 누수되지 않는다. -- 선택한 제한 초과 정책이 테스트와 문서에 일치한다. -- proto-socket transport 변경으로 문제를 우회하지 않았다. - -## 검증 결과 - -### REFACTOR-1 중간 검증 - -```bash -$ go test -race -count=1 ./apps/node/internal/node/... -(output) -``` - -### 최종 검증 - -```bash -$ go test -race -count=1 ./apps/node/internal/node/... -$ go test -count=1 ./apps/node/... -(output) -``` diff --git a/apps/node/internal/bootstrap/module.go b/apps/node/internal/bootstrap/module.go index c9c9923..2ef2a48 100644 --- a/apps/node/internal/bootstrap/module.go +++ b/apps/node/internal/bootstrap/module.go @@ -69,7 +69,8 @@ func Module(cfg *config.NodeConfig) fx.Option { } rtr := router.New(reg, logger) - n := node.New(result.NodeID, rtr, st, os.Stdout, logger) + globalConcurrency := int(result.Config.GetRuntime().GetConcurrency()) + n := node.New(result.NodeID, rtr, st, globalConcurrency, os.Stdout, logger) result.Session.SetEventHandler(func(event *iop.EdgeNodeEvent) { printEdgeEvent(os.Stdout, event) }) diff --git a/apps/node/internal/node/node.go b/apps/node/internal/node/node.go index 6aefc7a..4a083d4 100644 --- a/apps/node/internal/node/node.go +++ b/apps/node/internal/node/node.go @@ -14,6 +14,8 @@ import ( "strings" "time" + "sync" + "go.uber.org/zap" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" @@ -26,20 +28,27 @@ import ( // Node implements transport.Handler and coordinates the full execution pipeline. type Node struct { - nodeID string - router runtime.Router - store *store.Store - runs *runManager - out io.Writer - logger *zap.Logger + nodeID string + router runtime.Router + store *store.Store + runs *runManager + globalPermit *permitManager // node-wide limit across all adapters + adapterPermitsMu sync.Mutex + adapterPermits map[string]*permitManager // per adapter-key limit + out io.Writer + logger *zap.Logger } // New creates a Node. It satisfies transport.Handler. +// globalConcurrency is the node-wide maximum for simultaneous executions across +// all adapters (>= 1 enforces limit; <= 0 means unlimited at the node level). +// Per-adapter MaxConcurrency from Capabilities is applied independently on top. // out receives console debug output; pass os.Stdout for production, io.Discard in tests. func New( nodeID string, router runtime.Router, st *store.Store, + globalConcurrency int, out io.Writer, logger *zap.Logger, ) *Node { @@ -47,12 +56,14 @@ func New( out = os.Stdout } return &Node{ - nodeID: nodeID, - router: router, - store: st, - runs: newRunManager(), - out: out, - logger: logger, + nodeID: nodeID, + router: router, + store: st, + runs: newRunManager(), + globalPermit: newPermitManager(globalConcurrency), + adapterPermits: make(map[string]*permitManager), + out: out, + logger: logger, } } @@ -84,6 +95,42 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i return fmt.Errorf("node: resolve: %w", err) } + // Concurrency gate (foreground and background use the same path). + // Policy: reject-on-exceed (no queue, no wait). + // + // Two independent limits are enforced: + // 1. global (node-wide): n.globalPermit — shared across all adapter keys. + // 2. adapter-key: per-adapter Capabilities.MaxConcurrency permit. + // + // Both must be acquired. If global is acquired but adapter-key fails, global + // is immediately released. The reject limit reported to the caller is the + // stricter of the two. + adapterCap := 0 + if caps, capsErr := adapter.Capabilities(ctx); capsErr == nil { + adapterCap = caps.MaxConcurrency + } + adapterPM := n.adapterPermitFor(spec.Adapter, adapterCap) + + if !n.globalPermit.tryAcquire() { + limit := int(n.globalPermit.limit) + n.logger.Warn("global concurrency limit exceeded", + zap.String("run_id", spec.RunID), + zap.Int("global_limit", limit), + ) + n.rejectRun(ctx, sess, spec, limit) + return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded) + } + if !adapterPM.tryAcquire() { + n.globalPermit.release() + n.logger.Warn("adapter concurrency limit exceeded", + zap.String("run_id", spec.RunID), + zap.String("adapter", spec.Adapter), + zap.Int("adapter_limit", adapterCap), + ) + n.rejectRun(ctx, sess, spec, adapterCap) + return fmt.Errorf("node: run %s: %w", spec.RunID, ErrConcurrencyLimitExceeded) + } + if err := n.store.InsertRun(ctx, store.RunRecord{ RunID: spec.RunID, Adapter: spec.Adapter, @@ -120,6 +167,8 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i } run := func() error { + defer n.globalPermit.release() + defer adapterPM.release() defer cancel() defer n.runs.deregister(spec.RunID) defer close(h.done) @@ -136,6 +185,55 @@ func (n *Node) OnRunRequest(ctx context.Context, sess *transport.Session, req *i return run() } +// adapterPermitFor returns the shared permitManager for the given adapter +// instance key, creating one with the given limit if not yet seen. +// limit <= 0 means unlimited. +func (n *Node) adapterPermitFor(adapterKey string, limit int) *permitManager { + n.adapterPermitsMu.Lock() + defer n.adapterPermitsMu.Unlock() + pm, ok := n.adapterPermits[adapterKey] + if !ok { + pm = newPermitManager(limit) + n.adapterPermits[adapterKey] = pm + } + return pm +} + +// rejectRun records a rejected run in the store and sends an error RunEvent to +// the session so Edge can observe the rejection through the normal event stream. +func (n *Node) rejectRun(ctx context.Context, sess *transport.Session, spec runtime.ExecutionSpec, limit int) { + errMsg := fmt.Sprintf("concurrency limit exceeded (limit=%d)", limit) + + if err := n.store.InsertRun(ctx, store.RunRecord{ + RunID: spec.RunID, + Adapter: spec.Adapter, + Target: spec.Target, + SessionID: normalizeSessionID(spec.SessionID), + Background: spec.Background, + Status: "rejected", + CreatedAt: time.Now(), + }); err != nil { + n.logger.Warn("store: insert rejected run", zap.String("run_id", spec.RunID), zap.Error(err)) + } else if err := n.store.CompleteRun(ctx, spec.RunID, "rejected", errMsg); err != nil { + n.logger.Warn("store: complete rejected run", zap.String("run_id", spec.RunID), zap.Error(err)) + } + + if sess != nil && sess.IsAlive() { + re := &iop.RunEvent{ + RunId: spec.RunID, + Type: string(runtime.EventTypeError), + Error: errMsg, + Timestamp: time.Now().UnixNano(), + SessionId: normalizeSessionID(spec.SessionID), + Background: spec.Background, + NodeId: n.nodeID, + } + if err := sess.Send(re); err != nil { + n.logger.Warn("session: send reject event", zap.String("run_id", spec.RunID), zap.Error(err)) + } + } +} + func (n *Node) completeRun(spec runtime.ExecutionSpec, execErr error) { status := "completed" errMsg := "" diff --git a/apps/node/internal/node/node_concurrency_integration_test.go b/apps/node/internal/node/node_concurrency_integration_test.go new file mode 100644 index 0000000..7fa3f31 --- /dev/null +++ b/apps/node/internal/node/node_concurrency_integration_test.go @@ -0,0 +1,211 @@ +package node_test + +import ( + "context" + "fmt" + "net" + "testing" + "time" + + toki "git.toki-labs.com/toki/proto-socket/go" + "go.uber.org/zap" + "google.golang.org/protobuf/proto" + + "iop/apps/node/internal/node" + "iop/apps/node/internal/runtime" + "iop/apps/node/internal/store" + "iop/apps/node/internal/transport" + iop "iop/proto/gen/iop" +) + +// edgeServerParserMap returns the proto parser map for a mock Edge server +// that expects to receive RunEvent messages sent by Node. +func edgeServerParserMap() toki.ParserMap { + return toki.ParserMap{ + toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) { + m := &iop.RunEvent{} + return m, proto.Unmarshal(b, m) + }, + toki.TypeNameOf(&iop.RegisterRequest{}): func(b []byte) (proto.Message, error) { + m := &iop.RegisterRequest{} + return m, proto.Unmarshal(b, m) + }, + } +} + +func getFreeListenAddr(t *testing.T) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := l.Addr().String() + l.Close() + return addr +} + +// startMockEdge starts a mock Edge TCP server that accepts exactly one Node +// connection, performs the registration handshake, and returns the accepted +// toki.TcpClient so tests can attach listeners and send RunRequests. +func startMockEdge(t *testing.T, ctx context.Context, listenAddr string) *toki.TcpClient { + t.Helper() + + host, portStr, _ := net.SplitHostPort(listenAddr) + port := 0 + fmt.Sscanf(portStr, "%d", &port) + + acceptedCh := make(chan *toki.TcpClient, 1) + server := toki.NewTcpServer(host, port, func(conn net.Conn) *toki.TcpClient { + client := toki.NewTcpClient(conn, 30, 10, edgeServerParserMap()) + toki.AddRequestListenerTyped[*iop.RegisterRequest, *iop.RegisterResponse]( + &client.Communicator, + func(req *iop.RegisterRequest) (*iop.RegisterResponse, error) { + return &iop.RegisterResponse{ + Accepted: true, + NodeId: "test-node", + Alias: "test-alias", + Config: &iop.NodeConfigPayload{ + Runtime: &iop.NodeRuntimeConfig{Concurrency: 1}, + }, + }, nil + }, + ) + acceptedCh <- client + return client + }) + if err := server.Start(ctx); err != nil { + t.Fatalf("mock edge server start: %v", err) + } + t.Cleanup(func() { server.Stop() }) + + select { + case client := <-acceptedCh: + return client + case <-time.After(3 * time.Second): + t.Fatal("mock edge server did not accept connection within 3s") + return nil + } +} + +// TestConcurrencyLimit_RejectEventObservedByEdge is an integration test that +// verifies the RunEvent{type:"error"} sent by rejectRun() actually arrives at +// the mock Edge server when the Node's concurrency limit is exceeded. +// +// Setup: +// 1. Start mock Edge TCP server. +// 2. Dial Node-side session via transport.DialEdge. +// 3. Create a real node.Node with globalConcurrency=1. +// 4. Attach the session to the node via SetHandler. +// 5. From the Edge side, send a first RunRequest whose adapter blocks. +// 6. After the first run starts, send a second RunRequest from the Edge side. +// 7. Assert that a RunEvent{type:"error", run_id:} arrives at the Edge. +func TestConcurrencyLimit_RejectEventObservedByEdge(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + logger := zap.NewNop() + listenAddr := getFreeListenAddr(t) + + // Channel that the mock edge client will be delivered into once Node connects. + edgeClientCh := make(chan *toki.TcpClient, 1) + + // Start mock Edge server (async — we need to dial first to trigger accept). + host, portStr, _ := net.SplitHostPort(listenAddr) + port := 0 + fmt.Sscanf(portStr, "%d", &port) + server := toki.NewTcpServer(host, port, func(conn net.Conn) *toki.TcpClient { + client := toki.NewTcpClient(conn, 30, 10, edgeServerParserMap()) + toki.AddRequestListenerTyped[*iop.RegisterRequest, *iop.RegisterResponse]( + &client.Communicator, + func(req *iop.RegisterRequest) (*iop.RegisterResponse, error) { + return &iop.RegisterResponse{ + Accepted: true, + NodeId: "test-node", + Alias: "test-alias", + Config: &iop.NodeConfigPayload{ + Runtime: &iop.NodeRuntimeConfig{Concurrency: 1}, + }, + }, nil + }, + ) + edgeClientCh <- client + return client + }) + if err := server.Start(ctx); err != nil { + t.Fatalf("mock edge server start: %v", err) + } + defer server.Stop() + + // Dial Node side to Edge. + dialResult, err := transport.DialEdge(ctx, listenAddr, "test-token", logger) + if err != nil { + t.Fatalf("DialEdge: %v", err) + } + defer dialResult.Session.Close() + + // Wait for Edge side to accept the connection. + var edgeClient *toki.TcpClient + select { + case edgeClient = <-edgeClientCh: + case <-time.After(3 * time.Second): + t.Fatal("edge server did not accept connection") + } + + // Attach RunEvent listener on the Edge side before sending any requests. + runEventCh := make(chan *iop.RunEvent, 4) + toki.AddListenerTyped[*iop.RunEvent](&edgeClient.Communicator, func(event *iop.RunEvent) { + runEventCh <- event + }) + + // Build real node.Node with global concurrency=1. + sa := newSlowAdapter() // MaxConcurrency=1; blocks until sa.release is closed + rtr := &fixedRouter{ + adapterName: "slow", + adapters: map[string]runtime.Adapter{"slow": sa}, + } + st, err := store.New(":memory:", logger) + if err != nil { + t.Fatalf("store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + + n := node.New("test-node", rtr, st, 1, nil, logger) + dialResult.Session.SetHandler(n) + + // Send first RunRequest from Edge → Node (adapter will block). + firstReq := &iop.RunRequest{RunId: "run-hold-1", Adapter: "slow", Target: "v1"} + if err := edgeClient.Send(firstReq); err != nil { + t.Fatalf("send first run request: %v", err) + } + + // Wait until the first run's adapter has started executing so the permit is held. + select { + case <-sa.started: + case <-time.After(3 * time.Second): + t.Fatal("first run never started") + } + + // Send second RunRequest; should be rejected because global=1 is taken. + secondReq := &iop.RunRequest{RunId: "run-reject-2", Adapter: "slow", Target: "v1"} + if err := edgeClient.Send(secondReq); err != nil { + t.Fatalf("send second run request: %v", err) + } + + // Collect RunEvents until we see the error event for the second run. + deadline := time.After(5 * time.Second) + for { + select { + case ev := <-runEventCh: + if ev.GetRunId() == "run-reject-2" && ev.GetType() == string(runtime.EventTypeError) { + if ev.GetError() == "" { + t.Fatal("reject RunEvent has empty error field") + } + // Success: RunEvent{type:"error", run_id:"run-reject-2"} observed. + close(sa.release) + return + } + case <-deadline: + t.Fatal("timeout: did not receive RunEvent{type:error} for run-reject-2") + } + } +} diff --git a/apps/node/internal/node/node_test.go b/apps/node/internal/node/node_test.go index 7da9d35..0efc6aa 100644 --- a/apps/node/internal/node/node_test.go +++ b/apps/node/internal/node/node_test.go @@ -206,13 +206,18 @@ func (r *errorRouter) GetAdapter(_ string) (runtime.Adapter, bool) { } func makeNode(t *testing.T, rtr runtime.Router) (*node.Node, *store.Store) { + t.Helper() + return makeNodeWithConcurrency(t, rtr, 0) +} + +func makeNodeWithConcurrency(t *testing.T, rtr runtime.Router, concurrency int) (*node.Node, *store.Store) { t.Helper() st, err := store.New(":memory:", zap.NewNop()) if err != nil { t.Fatalf("store: %v", err) } t.Cleanup(func() { _ = st.Close() }) - return node.New("test-node", rtr, st, io.Discard, zap.NewNop()), st + return node.New("test-node", rtr, st, concurrency, io.Discard, zap.NewNop()), st } // --- existing tests (updated) --- @@ -917,3 +922,383 @@ func TestOnCommandRequest_Capabilities_ExactInstanceKey(t *testing.T) { t.Errorf("result[instance_key]: got %q want ollama@local", got) } } + +// --- concurrency gate tests --- + +// slowAdapter blocks until released, exposing a start channel so tests can +// synchronize. Execute returns nil on normal completion. +type slowAdapter struct { + started chan struct{} + release chan struct{} +} + +func newSlowAdapter() *slowAdapter { + return &slowAdapter{started: make(chan struct{}, 1), release: make(chan struct{})} +} + +func (a *slowAdapter) Name() string { return "slow" } +func (a *slowAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) { + return runtime.Capabilities{AdapterName: "slow", MaxConcurrency: 1}, nil +} + +// slowAdapterUnlimited is like slowAdapter but reports MaxConcurrency=0 (unlimited). +type slowAdapterUnlimited struct { + started chan struct{} + release chan struct{} +} + +func newSlowAdapterUnlimited() *slowAdapterUnlimited { + return &slowAdapterUnlimited{started: make(chan struct{}, 1), release: make(chan struct{})} +} + +func (a *slowAdapterUnlimited) Name() string { return "slow" } +func (a *slowAdapterUnlimited) Capabilities(_ context.Context) (runtime.Capabilities, error) { + return runtime.Capabilities{AdapterName: "slow", MaxConcurrency: 0}, nil +} +func (a *slowAdapterUnlimited) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error { + select { + case a.started <- struct{}{}: + default: + } + select { + case <-a.release: + case <-ctx.Done(): + return runtime.ErrRunCancelled + } + return nil +} + +// slowHighCapAdapter is like slowAdapter but reports MaxConcurrency=4, +// used to verify that the global limit overrides a looser adapter cap. +type slowHighCapAdapter struct { + started chan struct{} + release chan struct{} +} + +func newSlowHighCapAdapter() *slowHighCapAdapter { + return &slowHighCapAdapter{started: make(chan struct{}, 1), release: make(chan struct{})} +} + +func (a *slowHighCapAdapter) Name() string { return "slow-hicap" } +func (a *slowHighCapAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) { + return runtime.Capabilities{AdapterName: "slow-hicap", MaxConcurrency: 4}, nil +} +func (a *slowHighCapAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error { + select { + case a.started <- struct{}{}: + default: + } + select { + case <-a.release: + case <-ctx.Done(): + return runtime.ErrRunCancelled + } + return nil +} + +func (a *slowAdapter) Execute(ctx context.Context, _ runtime.ExecutionSpec, _ runtime.EventSink) error { + select { + case a.started <- struct{}{}: + default: + } + select { + case <-a.release: + case <-ctx.Done(): + return runtime.ErrRunCancelled + } + return nil +} + +// TestConcurrencyLimit_RejectSecondForeground verifies that a second foreground +// run is rejected immediately when the global limit is 1 and one run is active. +func TestConcurrencyLimit_RejectSecondForeground(t *testing.T) { + sa := newSlowAdapter() + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} + n, _ := makeNodeWithConcurrency(t, router, 1) + + errc := make(chan error, 1) + go func() { + errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-fg-1", Adapter: "slow", Target: "v1", + }) + }() + + // Wait until the first run has started executing. + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("first run never started") + } + + // Second run must be rejected. + err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-fg-2", Adapter: "slow", Target: "v1", + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err) + } + + // Release the first run and drain it. + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("first run: %v", err) + } +} + +// TestConcurrencyLimit_RejectBackgroundRun verifies that a background run is +// also subject to the global concurrency gate. +func TestConcurrencyLimit_RejectBackgroundRun(t *testing.T) { + sa := newSlowAdapter() + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} + n, _ := makeNodeWithConcurrency(t, router, 1) + + errc := make(chan error, 1) + go func() { + errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-bg-hold", Adapter: "slow", Target: "v1", + }) + }() + + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("hold run never started") + } + + // Background run must also be rejected. + err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-bg-2", Adapter: "slow", Target: "v1", Background: true, + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded for background run, got %v", err) + } + + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("hold run: %v", err) + } +} + +// TestConcurrencyLimit_PermitReleasedAfterCompletion verifies that after the +// first run completes, a subsequent run can acquire the permit. +func TestConcurrencyLimit_PermitReleasedAfterCompletion(t *testing.T) { + sa := newSlowAdapter() + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} + n, _ := makeNodeWithConcurrency(t, router, 1) + + // First run: start, wait, release. + errc := make(chan error, 1) + go func() { + errc <- n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-seq-1", Adapter: "slow", Target: "v1", + }) + }() + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("seq-1 never started") + } + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("seq-1: %v", err) + } + + // Second run must succeed now that the permit is free. + sa2 := newSlowAdapter() + router.adapters["slow"] = sa2 + close(sa2.release) // let it complete immediately + if err := n.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-seq-2", Adapter: "slow", Target: "v1", + }); err != nil { + t.Fatalf("seq-2 should succeed after permit released, got %v", err) + } +} + +// TestConcurrencyLimit_Unlimited verifies that global=0 + adapter cap=0 imposes no cap. +func TestConcurrencyLimit_Unlimited(t *testing.T) { + const count = 5 + shared := newSlowAdapterUnlimited() // MaxConcurrency=0 (unlimited) + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": shared}} + nd, _ := makeNodeWithConcurrency(t, router, 0) // global unlimited + + errs := make(chan error, count) + for i := 0; i < count; i++ { + runID := fmt.Sprintf("run-unlimited-%d", i) + go func(id string) { + errs <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: id, Adapter: "slow", Target: "v1", + }) + }(runID) + } + + // Release all. + close(shared.release) + + // All must complete without ErrConcurrencyLimitExceeded. + for i := 0; i < count; i++ { + if err := <-errs; errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("unexpected concurrency rejection with limit=0: %v", err) + } + } +} + +// TestConcurrencyLimit_AdapterCapFallback verifies that when global=0 and adapter +// cap=1, the adapter cap is used as the effective limit. +func TestConcurrencyLimit_AdapterCapFallback(t *testing.T) { + sa := newSlowAdapter() // MaxConcurrency=1 + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} + nd, _ := makeNodeWithConcurrency(t, router, 0) // global=0 → use adapter cap + + errc := make(chan error, 1) + go func() { + errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-cap-1", Adapter: "slow", Target: "v1", + }) + }() + + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("run-cap-1 never started") + } + + // Second run must be rejected by adapter cap. + err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-cap-2", Adapter: "slow", Target: "v1", + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded from adapter cap, got %v", err) + } + + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("run-cap-1: %v", err) + } +} + +// TestConcurrencyLimit_GlobalBeatsAdapterCap verifies that when both global and +// adapter cap are > 0 and global < adapter cap, the global (stricter) limit applies. +func TestConcurrencyLimit_GlobalBeatsAdapterCap(t *testing.T) { + // slowHighCapAdapter reports MaxConcurrency=4; global=1 → effective=1. + sa := newSlowHighCapAdapter() + router := &fixedRouter{adapterName: "slow-hicap", adapters: map[string]runtime.Adapter{"slow-hicap": sa}} + nd, _ := makeNodeWithConcurrency(t, router, 1) // global=1, adapter cap=4 → effective=1 + + errc := make(chan error, 1) + go func() { + errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-g1-1", Adapter: "slow-hicap", Target: "v1", + }) + }() + + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("run-g1-1 never started") + } + + // Second run must be rejected by global limit=1 even though adapter cap=4. + err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-g1-2", Adapter: "slow-hicap", Target: "v1", + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded (global=1 beats adapter cap=4), got %v", err) + } + + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("run-g1-1: %v", err) + } +} + +// TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters verifies that +// global=1 rejects a run on adapter B when adapter A already holds the global +// permit, even though both adapter caps are > 1. +func TestConcurrencyLimit_GlobalBeatsAdapterCap_DifferentAdapters(t *testing.T) { + saA := newSlowHighCapAdapter() // adapter "slow-hicap-a", MaxConcurrency=4 + saB := newSlowHighCapAdapter() // adapter "slow-hicap-b", MaxConcurrency=4 + + router := &fixedRouter{ + adapterName: "slow-hicap-a", + adapters: map[string]runtime.Adapter{ + "slow-hicap-a": saA, + "slow-hicap-b": saB, + }, + } + nd, _ := makeNodeWithConcurrency(t, router, 1) // global=1 + + // First run on adapter A: hold the global permit. + errc := make(chan error, 1) + go func() { + errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-xa-1", Adapter: "slow-hicap-a", Target: "v1", + }) + }() + select { + case <-saA.started: + case <-time.After(2 * time.Second): + t.Fatal("run-xa-1 never started") + } + + // Second run on adapter B must be rejected because global permit is taken. + router.adapterName = "slow-hicap-b" + err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-xb-1", Adapter: "slow-hicap-b", Target: "v1", + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded for cross-adapter second run, got %v", err) + } + + close(saA.release) + if err := <-errc; err != nil { + t.Fatalf("run-xa-1: %v", err) + } +} + +// TestConcurrencyLimit_RejectStoreAndEvent verifies that a rejected run is stored +// with terminal status "rejected" and an error message is recorded. +func TestConcurrencyLimit_RejectStoreAndEvent(t *testing.T) { + sa := newSlowAdapter() + router := &fixedRouter{adapterName: "slow", adapters: map[string]runtime.Adapter{"slow": sa}} + nd, st := makeNodeWithConcurrency(t, router, 1) + + errc := make(chan error, 1) + go func() { + errc <- nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-hold", Adapter: "slow", Target: "v1", + }) + }() + select { + case <-sa.started: + case <-time.After(2 * time.Second): + t.Fatal("run-hold never started") + } + + // Second run must be rejected. + err := nd.OnRunRequest(context.Background(), &transport.Session{}, &iop.RunRequest{ + RunId: "run-rejected", Adapter: "slow", Target: "v1", + }) + if !errors.Is(err, node.ErrConcurrencyLimitExceeded) { + t.Fatalf("expected ErrConcurrencyLimitExceeded, got %v", err) + } + + // Rejected run must be stored with status "rejected". + run, err := st.GetRun(context.Background(), "run-rejected") + if err != nil { + t.Fatalf("GetRun: %v", err) + } + if run == nil { + t.Fatal("expected rejected run to be stored") + } + if run.Status != "rejected" { + t.Fatalf("expected status %q, got %q", "rejected", run.Status) + } + if run.Error == "" { + t.Fatal("expected non-empty error message for rejected run") + } + + close(sa.release) + if err := <-errc; err != nil { + t.Fatalf("run-hold: %v", err) + } +} diff --git a/apps/node/internal/node/run_manager.go b/apps/node/internal/node/run_manager.go index b85eb1f..c275b19 100644 --- a/apps/node/internal/node/run_manager.go +++ b/apps/node/internal/node/run_manager.go @@ -2,9 +2,15 @@ package node import ( "context" + "errors" "sync" + "sync/atomic" ) +// ErrConcurrencyLimitExceeded is returned when a run is rejected because the +// active run count has reached the effective concurrency limit. +var ErrConcurrencyLimitExceeded = errors.New("concurrency limit exceeded") + type runHandle struct { runID string adapter string @@ -45,3 +51,42 @@ func (m *runManager) cancelRun(runID string) bool { h.cancel() return true } + +// permitManager enforces a concurrency limit using an atomic counter. +// limit <= 0 means unlimited. Policy is reject-on-exceed (no queue, no wait). +type permitManager struct { + limit int32 + active atomic.Int32 +} + +func newPermitManager(limit int) *permitManager { + return &permitManager{limit: int32(limit)} +} + +// tryAcquire attempts to take a permit. Returns false immediately if the limit +// is already reached. limit <= 0 always succeeds. +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 + } + } +} + +// release returns a permit. Must be called exactly once per successful tryAcquire. +func (p *permitManager) release() { + p.active.Add(-1) +} + +// activeCount returns the number of currently held permits (for observability). +func (p *permitManager) activeCount() int { + return int(p.active.Load()) +} diff --git a/edge b/edge new file mode 100755 index 0000000..628863c Binary files /dev/null and b/edge differ diff --git a/node b/node new file mode 100755 index 0000000..e53e792 Binary files /dev/null and b/node differ