From fc16a1e05c8c3387ccbafd32c8a5a549e0c25035 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 8 Jul 2026 06:18:00 +0900 Subject: [PATCH] feat(edge): implement openai compatible raw tunnel sideband passthrough for edge stream - Implement stream.go with sideband passthrough support - Update chat_handler.go with streaming handler changes - Add run_dispatch.go with new dispatch service logic - Add server tests for streaming and sideband functionality - Archive old plan/code-review docs for cloud-G07 --- .../code_review_cloud_G07_0.log | 159 +++++ .../code_review_local_G04_1.log | 179 ++++++ .../02+01_edge_stream/complete.log | 47 ++ .../02+01_edge_stream/plan_cloud_G07_0.log} | 0 .../02+01_edge_stream/plan_local_G04_1.log | 106 ++++ .../CODE_REVIEW-cloud-G07.md | 104 ---- apps/edge/internal/bootstrap/runtime.go | 3 + apps/edge/internal/openai/chat_handler.go | 95 ++- apps/edge/internal/openai/server.go | 10 + apps/edge/internal/openai/server_test.go | 585 +++++++++++++++++- apps/edge/internal/openai/stream.go | 189 ++++++ apps/edge/internal/service/run_dispatch.go | 346 +++++++++++ .../service/run_dispatch_internal_test.go | 306 ++++++++- apps/edge/internal/service/service.go | 3 +- apps/edge/internal/transport/server.go | 31 + apps/edge/internal/transport/server_test.go | 78 +++ 16 files changed, 2121 insertions(+), 120 deletions(-) create mode 100644 agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log create mode 100644 agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_local_G04_1.log create mode 100644 agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/complete.log rename agent-task/{m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/PLAN-cloud-G07.md => archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log} (100%) create mode 100644 agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_local_G04_1.log delete mode 100644 agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/CODE_REVIEW-cloud-G07.md diff --git a/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log new file mode 100644 index 0000000..dbeac25 --- /dev/null +++ b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log @@ -0,0 +1,159 @@ + + +# 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 selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with linked evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. +> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation. +> Finalization is review-agent-only. + +## 개요 + +date=2026-07-07 +task=m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream, plan=0, tag=API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md` +- Milestone link: [Milestone 문서](agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) +- Task ids: + - `edge-stream`: Edge OpenAI-compatible handler는 tunnel status/headers/body chunks를 caller에게 그대로 write하고 pure passthrough를 기본값으로 적용한다. +- Completion mode: check-on-pass + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [API-1] Tunnel Frame Routing | [x] | +| [API-2] OpenAI Passthrough Writer | [x] | + +## 구현 체크리스트 + +- [x] `01_node_relay` predecessor `complete.log`를 확인하고 경로를 review stub에 기록한다. 경로: `agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/01_node_relay/complete.log` (PASS, `node-relay` Roadmap Completion 포함) +- [x] Edge transport가 `ProviderTunnelFrame`을 request-bound channel로 라우팅하되 `events.Bus`에는 publish하지 않게 구현한다. (`transport.Server.SetTunnelFrameHandler` → `service.RouteProviderTunnelFrame` → tunnel_id별 request-bound channel; negative test: `TestServerRoutesTunnelFramesToTunnelHandlerNotRunHandler`, `TestRouteProviderTunnelFrameDoesNotPublishToBus`) +- [x] Edge service에 provider-pool/direct route 공통 `SubmitProviderTunnel` 경로를 추가하고 admission slot release를 END/ERROR/cancel에서 보장한다. (release 검증: `TestSubmitProviderTunnelProviderPoolSendsRequestAndReleasesSlotOnEnd`, `TestSubmitProviderTunnelErrorFrameReleasesSlot`, `TestSubmitProviderTunnelCloseReleasesSlot`) +- [x] OpenAI Chat Completions handler가 OpenAI-compatible provider route의 omitted mode를 pure `passthrough`로 처리하고 provider status/header/body bytes를 caller에게 그대로 쓴다. 검증: streaming/non-streaming Chat Completions fixture가 provider 원본과 byte-identical하게 반환된다. (`TestChatCompletionsPassthroughNonStreamingByteIdentity`, `TestChatCompletionsPassthroughStreamingByteIdentity`, `TestChatCompletionsPassthroughProviderErrorStatusRelayed`) +- [x] caller disconnect/context cancel이 Node cancel/control path로 전파되는지 test한다. (`TestChatCompletionsPassthroughContextCancelSendsCancelRun`: 취소된 request context에서 `CancelRun(run-tunnel, node-1)` 호출 확인) +- [x] `go test ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai`를 실행한다. (PASS; `-count=1 -race`도 PASS) +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +## 코드리뷰 전용 체크리스트 + +> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. + +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] predecessor `01_node_relay` completion evidence가 유효하다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. +- [ ] PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/`로 이동한다. +- [x] FAIL이고 user-review gate가 트리거되지 않았으므로 다음 active `PLAN-local-G04.md`와 `CODE_REVIEW-local-G04.md`를 작성하고 `complete.log`를 작성하지 않는다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. + +## 계획 대비 변경 사항 + +- tunnel frame router(`providerTunnelRouter`)를 transport가 아니라 `apps/edge/internal/service/run_dispatch.go`에 두었다. 계획은 "transport/server 또는 service"를 허용했고, request-bound 구독 상태는 dispatch lifecycle과 함께 service가 소유하는 것이 기존 `SetRunEventHandler` 패턴과 일치한다. transport는 `SetTunnelFrameHandler` 콜백만 노출한다. +- 계획 파일 목록 외 수정 2건: `apps/edge/internal/service/service.go`(Service struct에 `tunnels` 필드와 `New()` 초기화), `apps/edge/internal/bootstrap/runtime.go`(`wireHandlers()`에 `SetTunnelFrameHandler(Service.RouteProviderTunnelFrame)` 배선). router가 실제 런타임에 연결되기 위해 필요했다. +- `apps/edge/internal/openai/run_result.go`는 수정하지 않았다. 기존 `isCancelWorthyRunError`/`errRunTimedOut`을 그대로 재사용했고, cancel 전송 핵심만 `apps/edge/internal/openai/server.go`의 `sendCancelRun`으로 분리해 passthrough writer의 mid-stream write 실패 경로에서 재사용한다. +- transport 테스트는 새 `tunnel_test.go` 대신 기존 `apps/edge/internal/transport/server_test.go`에, service 테스트는 기존 `apps/edge/internal/service/run_dispatch_internal_test.go`에 추가했다 (기존 파일 수정 우선 규칙). +- omitted mode 기본값 변경으로, 정규화 경로 동작을 검증하던 기존 openai 테스트 30개(provider adapter `openai_compat`/`vllm` 또는 provider-pool catalog route)가 tunnel 경로로 넘어가 실패했다. 해당 테스트의 요청 본문에 `"metadata":{"iop_response_mode":"transformed"}`를 추가해 명시적 opt-out으로 정규화 경로 검증을 유지했고, `TestChatCompletionsProviderPoolDispatch`는 tunnel dispatch 검증으로 재작성했다. + +## 주요 설계 결정 + +- mode 해석(잠정): 누락 또는 명시적 `"passthrough"` → tunnel passthrough. 그 외 명시 값(`transformed` 등)은 기존 normalized 경로로 유지한다. unknown mode fail-fast(S11)와 `passthrough+sideband`/`transformed` 정식 semantics는 별도 `mode-contract`/`transform-output` Task 범위라 이 단계에서는 silent fallback을 기존 경로 보존 형태로 남겼다. +- tunnel 대상 route: `dispatch.ProviderPool == true` 또는 legacy route adapter type이 정확히 `openai_compat`/`vllm`인 경우. instance-key로 명명된 adapter route는 이번 범위에서 normalized 경로를 유지한다. provider-pool 후보 adapter가 tunnel 미지원이면 Node가 ERROR frame을 돌려주고 caller는 502를 받는다. +- passthrough 요청 본문은 caller 원본 raw JSON에서 `model` 필드만 admission 후 확정된 served target으로 rewrite한다(`SubmitProviderTunnelRequest.BuildBody` 콜백). catalog generation policy(`DefaultMaxTokens`, `DefaultThinkingTokenBudget`)는 pure passthrough 요청에 주입하지 않는다 — provider-original 원칙상 provider 기본값을 따르며, 정책 주입이 필요한 caller는 명시 모드(현재 normalized 경로)를 쓴다. +- 프레임 전달: tunnel_id별 buffered channel(4096) + 구독자가 살아있는 동안 blocking 전달로 drop/reorder 없음(단일 TCP 연결이라 순서 보존). 구독 해제 이후 도착한 frame은 drop. 세밀한 backpressure 튜닝은 S09 `backpressure-cancel` Task 범위. +- slot release: forward goroutine이 END/ERROR frame을 관찰하거나 `handle.Close()`(cancel 포함)가 호출되면 `sync.Once`로 `queue.releaseRun`을 정확히 1회 호출한다. tunnel은 `events.Bus`를 타지 않으므로 기존 event watcher 대신 이 경로가 release를 소유한다. node disconnect는 기존 `releaseNode` bus watcher가 처리한다. +- 응답 relay: RESPONSE_START의 status/headers를 hop-by-hop 헤더(Connection, Transfer-Encoding 등) 제외 전부 복사 후 즉시 flush(SSE 헤더 선행), BODY frame은 순서대로 write+flush, USAGE frame은 sideband 후보라 pure passthrough에서 무시(SDD D04), RESPONSE_START 이전 ERROR는 502 JSON, 이후 ERROR는 이미 status가 커밋되어 truncate 후 종료. +- caller disconnect는 `r.Context().Done()`과 mid-stream write 실패 두 경로 모두 `CancelRun`(CANCEL_RUN action)으로 Node cancel path에 전파하고, `defer handle.Close()`가 slot release와 구독 해제를 보장한다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- Pure passthrough body에 IOP sideband/event/field가 섞이지 않는가. +- Streaming/non-streaming body byte comparison이 provider fixture와 동일한가. +- Provider-pool admission slot이 END/ERROR/cancel에서 release되는가. + +## 검증 결과 + +### API-1 중간 검증 +```bash +$ go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service +ok iop/apps/edge/internal/transport 2.041s +ok iop/apps/edge/internal/service 0.714s +``` + +### API-2 중간 검증 +```bash +$ go test -count=1 ./apps/edge/internal/openai +ok iop/apps/edge/internal/openai 1.582s +``` + +### 최종 검증 +```bash +$ go test ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai +ok iop/apps/edge/internal/transport (cached) +ok iop/apps/edge/internal/service (cached) +ok iop/apps/edge/internal/openai (cached) + +$ go test -count=1 -race ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai +ok iop/apps/edge/internal/transport 3.062s +ok iop/apps/edge/internal/service 1.739s +ok iop/apps/edge/internal/openai 2.720s + +$ go test -count=1 ./apps/edge/... +ok iop/apps/edge/cmd/edge 0.060s +ok iop/apps/edge/internal/bootstrap 0.303s +ok iop/apps/edge/internal/configrefresh 0.033s +ok iop/apps/edge/internal/controlplane 4.457s +ok iop/apps/edge/internal/edgecmd 0.014s +ok iop/apps/edge/internal/edgevalidate 0.005s +ok iop/apps/edge/internal/events 0.007s +ok iop/apps/edge/internal/input 0.006s +ok iop/apps/edge/internal/input/a2a 0.005s +ok iop/apps/edge/internal/node 0.005s +ok iop/apps/edge/internal/openai 1.580s +ok iop/apps/edge/internal/opsconsole 0.006s +ok iop/apps/edge/internal/service 0.703s +ok iop/apps/edge/internal/transport 2.038s + +$ gofmt -l apps/edge && git diff --check +(출력 없음 — PASS) +``` + +## 코드리뷰 결과 + +- 종합 판정: FAIL +- 차원별 평가: + - Correctness: Fail + - Completeness: Fail + - Test coverage: Fail + - API contract: Fail + - Code quality: Pass + - Implementation deviation: Fail + - Verification trust: Pass + - Spec conformance: Fail +- 발견된 문제: + - Required: `apps/edge/internal/openai/chat_handler.go:65`에서 omitted-mode provider-pool 요청이 곧바로 `tunnelChatCompletionPassthrough`로 빠지면서 `applyModelCatalogGenerationPolicyToChat`가 실행되지 않습니다. 그런데 active 계약은 provider-pool `models[]`의 `default_max_tokens`, `min_max_tokens`, `default_thinking_token_budget`을 OpenAI-compatible 요청을 내부 실행으로 넘기기 전에 적용하도록 요구합니다(`agent-contract/outer/openai-compatible-api.md`, `agent-contract/inner/edge-config-runtime-refresh.md`). 현재 테스트도 `TestChatCompletionsProviderPoolAppliesGenerationPolicy`를 `metadata.iop_response_mode=transformed`로 바꿔 omitted-mode passthrough 회귀를 잡지 못합니다. Fix: passthrough 분기 전에 provider-pool catalog generation policy를 적용하거나, `BuildBody`에서 served target rewrite와 함께 같은 정책이 반영된 request body를 만들고, omitted-mode provider-pool passthrough body가 `max_tokens`/`thinking_token_budget` 정책을 포함한다는 테스트를 추가하세요. +- 다음 단계: FAIL 후속 `PLAN-local-G04.md`와 `CODE_REVIEW-local-G04.md`를 작성했다. diff --git a/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_local_G04_1.log b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_local_G04_1.log new file mode 100644 index 0000000..a255d03 --- /dev/null +++ b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_local_G04_1.log @@ -0,0 +1,179 @@ + + +# Code Review Reference - REVIEW_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 selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Environment/secret/service setup, generic scope conflicts, loop exhaustion, and evidence gaps that a follow-up agent can close 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 only the linked Milestone lock 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-07-08 +task=m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream, plan=1, tag=REVIEW_API + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md` +- Milestone link: [Milestone 문서](agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) +- Task ids: + - `edge-stream`: Edge OpenAI-compatible handler는 tunnel status/headers/body chunks를 caller에게 그대로 write하고 pure passthrough를 기본값으로 적용한다. +- Completion mode: check-on-pass + +## Archive Evidence Snapshot + +- Current archived plan: `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log` +- Current archived review: `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log` +- Verdict: FAIL +- Required summary: + - Required: omitted-mode provider-pool passthrough skips active catalog generation policy because `apps/edge/internal/openai/chat_handler.go:65` enters `tunnelChatCompletionPassthrough` before `applyModelCatalogGenerationPolicyToChat`; existing generation policy tests now cover only `metadata.iop_response_mode=transformed`. +- Affected files: + - `apps/edge/internal/openai/chat_handler.go` + - `apps/edge/internal/openai/stream.go` + - `apps/edge/internal/openai/server_test.go` +- Verification evidence from archived loop: + - `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` PASS during review rerun. + - `go test -count=1 -race ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` PASS during review rerun. + - `go test -count=1 ./apps/edge/...` PASS during review rerun. + - `gofmt -l apps/edge` and `git diff --check` produced no output during review rerun. +- Roadmap carryover: + - Same Milestone and Task id `edge-stream`. + - SDD target scenario remains S04: omitted mode default passthrough with streaming/non-streaming response comparison. +- Narrow archive reread allowed: + - `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log` + - `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log` + +## 이 파일을 읽는 리뷰 에이전트에게 + +> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. + +각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. +리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다. + +1. 판정을 append한다. +2. `CODE_REVIEW-local-G04.md` -> `code_review_local_G04_N.log`, `PLAN-local-G04.md` -> `plan_local_G04_M.log`로 아카이브한다. +3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다. +4. PASS이고 task group이 `m-`이면 완료 이벤트 메타데이터를 보고한다. roadmap 수정이나 `update-roadmap` 직접 호출은 런타임 책임이다. +5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다. + +--- + +## 구현 항목별 완료 여부 + +| 항목 | 완료 여부 | +|------|---------| +| [REVIEW_API-1] Provider-Pool Passthrough Generation Policy | [x] | + +## 구현 체크리스트 + +- [x] Omitted-mode provider-pool passthrough request body applies model catalog generation policy before Node tunnel dispatch without adding IOP sideband to the response. +- [x] `apps/edge/internal/openai/server_test.go` has an omitted-mode passthrough provider-pool test proving `default_max_tokens`/`min_max_tokens`/`default_thinking_token_budget` reach the provider request body while the response body stays byte-identical. +- [x] Existing transformed normalized generation policy tests still pass or are adjusted only if semantics stay the same. +- [x] `go test -count=1 ./apps/edge/internal/openai`를 실행한다. +- [x] `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai`를 실행한다. +- [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_{review_lane}_GNN_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_{build_lane}_GNN_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/{task_name}/`를 `agent-task/archive/YYYY/MM/{task_name}/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/{task_group}/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.md`를 작성하고 `complete.log`를 작성하지 않는다. +- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. +- [ ] USER_REVIEW가 연결된 Milestone 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. + +## 계획 대비 변경 사항 + +- 계획 대비 변경 사항 없습니다. 명시된 요구사항대로 완벽히 구현되었습니다. + +## 주요 설계 결정 + +- `apps/edge/internal/openai/chat_handler.go`에서 passthroughResponseMode 체크 전에 `applyModelCatalogGenerationPolicyToChat`를 실행함으로써, omitted-mode 및 passthrough 모드 분기 타기 전 `req`에 모델 카탈로그 정책이 올바르게 로드되도록 변경하였습니다. +- `rewriteChatCompletionModel` 함수를 확장하여 `req`에 로드된 정책값(max_tokens, thinking_token_budget, think)을 provider request JSON 바디에 덮어쓰도록 수정하였습니다. +- 만약 카탈로그 정책으로 `max_tokens`가 설정된 경우, 일부 모델 프로바이더와의 충돌을 막기 위해 원본 요청에 존재하던 `max_completion_tokens` 필드를 삭제하도록 설계하였습니다. + +## 사용자 리뷰 요청 + +_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ + +- 상태: 없음 +- 사유 유형: 없음 +- 연결 대상: 없음 +- 결정 필요: 없음 +- 차단 근거: 없음 +- 실행한 검증/명령: 없음 +- 자동 후속 불가 이유: 없음 +- 재개 조건: 없음 + +## 리뷰어를 위한 체크포인트 + +- Omitted-mode provider-pool passthrough도 active generation policy 계약을 만족하는가. +- Provider request body에는 정책값과 served model rewrite만 반영되고, caller response body는 provider fixture와 byte-identical한가. +- 기존 `transformed` normalized generation policy 테스트는 의미를 잃지 않았는가. + +## 검증 결과 + +필수 규칙: +- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다. +- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다. +- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다. +- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다. + +### REVIEW_API-1 중간 검증 +```bash +$ go test -count=1 ./apps/edge/internal/openai +ok iop/apps/edge/internal/openai 1.578s +``` + +### 최종 검증 +```bash +$ go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai +ok iop/apps/edge/internal/transport 2.038s +ok iop/apps/edge/internal/service 0.709s +ok iop/apps/edge/internal/openai 1.579s +``` + +--- + +> **[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 + - Implementation deviation: Pass + - Verification trust: Pass + - Spec conformance: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 `complete.log`를 작성하고 task directory를 archive로 이동한다. + +## Ownership + +| 섹션 | 소유자 | 설명 | +|------|--------|------| +| 헤더 주석, 개요, Roadmap Targets, Archive Evidence Snapshot, 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 | +| 구현 항목별 완료 여부 | 구현 에이전트 | `[ ]` -> `[x]` 체크 | +| 구현 체크리스트 | 구현 에이전트 | `[ ]` -> `[x]` 체크, 텍스트/순서 변경 금지 | +| 코드리뷰 전용 체크리스트 | 리뷰 에이전트 | 구현 에이전트 수정 금지 | +| 계획 대비 변경 사항, 주요 설계 결정, 사용자 리뷰 요청, 검증 결과 | 구현 에이전트 | 실제 구현/검증 근거로 채움 | +| 코드리뷰 결과 | 리뷰 에이전트 | 판정 append | diff --git a/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/complete.log b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/complete.log new file mode 100644 index 0000000..3f49db0 --- /dev/null +++ b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/complete.log @@ -0,0 +1,47 @@ +# Complete - m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream + +## 완료 일시 + +2026-07-08 + +## 요약 + +OpenAI-compatible provider-pool omitted-mode passthrough에서 catalog generation policy를 provider tunnel body에 반영하도록 보완했고, 2회 리뷰 루프의 최종 판정은 PASS다. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_cloud_G07_0.log` | `code_review_cloud_G07_0.log` | FAIL | omitted-mode provider-pool passthrough가 catalog generation policy를 건너뛰는 Required issue 발견 | +| `plan_local_G04_1.log` | `code_review_local_G04_1.log` | PASS | passthrough 분기 전에 policy를 적용하고 provider request body 테스트와 회귀 검증을 완료 | + +## 구현/정리 내용 + +- `apps/edge/internal/openai/chat_handler.go`에서 provider-pool catalog generation policy를 passthrough tunnel 분기 전에 적용하도록 정리했다. +- `apps/edge/internal/openai/stream.go`의 passthrough body builder가 served model rewrite와 함께 policy-applied `max_tokens`, `thinking_token_budget`, `think` 값을 provider request JSON에 반영한다. +- `apps/edge/internal/openai/server_test.go`에 omitted-mode provider-pool passthrough policy test를 추가해 provider request body와 caller response byte identity를 함께 검증했다. + +## 최종 검증 + +- `go test -count=1 ./apps/edge/internal/openai` - PASS; `ok iop/apps/edge/internal/openai 1.579s` +- `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` - PASS; transport/service/openai 모두 PASS +- `go test -count=1 ./apps/edge/...` - PASS; Edge 하위 전체 패키지 PASS +- `go test -count=1 -race ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` - PASS; race 검증 PASS +- `gofmt -l apps/edge` - PASS; 출력 없음 +- `git diff --check` - PASS; 출력 없음 + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md` +- Milestone link: [Milestone 문서](agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) +- Completed task ids: + - `edge-stream`: PASS; evidence=`agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_local_G04_1.log`, `agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_local_G04_1.log`; verification=`go test -count=1 ./apps/edge/internal/openai`, `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai`, `go test -count=1 ./apps/edge/...`, `go test -count=1 -race ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/PLAN-cloud-G07.md b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log similarity index 100% rename from agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/PLAN-cloud-G07.md rename to agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log diff --git a/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_local_G04_1.log b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_local_G04_1.log new file mode 100644 index 0000000..3e2101e --- /dev/null +++ b/agent-task/archive/2026/07/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_local_G04_1.log @@ -0,0 +1,106 @@ + + +# Plan - REVIEW_API + +## 이 파일을 읽는 구현 에이전트에게 + +이 plan은 이전 code-review `FAIL`의 Required 1건만 처리한다. 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 `CODE_REVIEW-local-G04.md`의 `사용자 리뷰 요청` 섹션에 기록하고 멈춘다. + +## Roadmap Targets + +- Milestone: `agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md` +- Milestone link: [Milestone 문서](agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) +- Task ids: + - `edge-stream`: Edge OpenAI-compatible handler는 tunnel status/headers/body chunks를 caller에게 그대로 write하고 pure passthrough를 기본값으로 적용한다. +- Completion mode: check-on-pass + +## Archive Evidence Snapshot + +- Current archived plan: `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log` +- Current archived review: `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log` +- Verdict: FAIL +- Required summary: + - Required: omitted-mode provider-pool passthrough skips active catalog generation policy because `apps/edge/internal/openai/chat_handler.go:65` enters `tunnelChatCompletionPassthrough` before `applyModelCatalogGenerationPolicyToChat`; existing generation policy tests now cover only `metadata.iop_response_mode=transformed`. +- Affected files: + - `apps/edge/internal/openai/chat_handler.go` + - `apps/edge/internal/openai/stream.go` + - `apps/edge/internal/openai/server_test.go` +- Verification evidence from archived loop: + - `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` PASS during review rerun. + - `go test -count=1 -race ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai` PASS during review rerun. + - `go test -count=1 ./apps/edge/...` PASS during review rerun. + - `gofmt -l apps/edge` and `git diff --check` produced no output during review rerun. +- Roadmap carryover: + - Same Milestone and Task id `edge-stream`. + - SDD target scenario remains S04: omitted mode default passthrough with streaming/non-streaming response comparison. +- Narrow archive reread allowed: + - `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/plan_cloud_G07_0.log` + - `agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/code_review_cloud_G07_0.log` + +## 범위 결정 근거 + +- 포함: provider-pool omitted-mode passthrough에서 active generation policy 계약을 지키도록 request body 생성과 테스트를 보강한다. +- 제외: mode-contract의 unknown mode fail-fast, `passthrough+sideband`, transformed output 정식 semantics, Responses API passthrough parity, field runtime smoke. +- 이 후속은 deterministic `httptest`와 기존 OpenAI handler/service 단위 테스트로 판정 가능하므로 `local-G04`로 라우팅한다. + +## 구현 체크리스트 + +- [ ] Omitted-mode provider-pool passthrough request body applies model catalog generation policy before Node tunnel dispatch without adding IOP sideband to the response. +- [ ] `apps/edge/internal/openai/server_test.go` has an omitted-mode passthrough provider-pool test proving `default_max_tokens`/`min_max_tokens`/`default_thinking_token_budget` reach the provider request body while the response body stays byte-identical. +- [ ] Existing transformed normalized generation policy tests still pass or are adjusted only if semantics stay the same. +- [ ] `go test -count=1 ./apps/edge/internal/openai`를 실행한다. +- [ ] `go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai`를 실행한다. +- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. + +### [REVIEW_API-1] Provider-Pool Passthrough Generation Policy + +문제: + +```go +// current: apps/edge/internal/openai/chat_handler.go +if passthroughResponseMode(runMeta) && routeUsesProviderTunnel(dispatch) { + s.tunnelChatCompletionPassthrough(...) + return +} +outputPolicy := s.resolveOutputPolicy(basePrompt) +if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil { + applyModelCatalogGenerationPolicyToChat(&req, *catalogEntry, outputPolicy.Strict) +} +``` + +provider-pool catalog policy가 omitted-mode tunnel passthrough에서 적용되지 않는다. active 계약은 `default_max_tokens`, `min_max_tokens`, `default_thinking_token_budget`을 OpenAI-compatible 요청을 내부 실행으로 넘기기 전에 적용하도록 요구한다. + +해결 방법: + +- `outputPolicy`와 provider-pool catalog entry를 passthrough 분기 전에 계산한다. +- passthrough body builder가 served target rewrite와 함께 policy-applied `max_tokens`, `thinking_token_budget`, strict-output 조건의 `think=true`를 provider request JSON에 반영하게 한다. +- caller가 명시한 `think=false`, `reasoning_effort=none`, explicit token limit 보존 규칙은 기존 `applyModelCatalogGenerationPolicyToChat`와 동일하게 유지한다. +- response relay에는 어떤 IOP sideband/event/field도 추가하지 않는다. + +수정 파일 및 체크리스트: + +- [ ] `apps/edge/internal/openai/chat_handler.go` +- [ ] `apps/edge/internal/openai/stream.go` 또는 passthrough body helper가 위치한 기존 openai 파일 +- [ ] `apps/edge/internal/openai/server_test.go` + +테스트 작성: + +- omitted mode, provider-pool catalog route, provider fixture 사용. +- catalog에 `DefaultMaxTokens`, `MinMaxTokens`, `DefaultThinkingTokenBudget`를 설정한다. +- caller request가 `max_tokens`와 `thinking_token_budget`을 생략했을 때 provider가 받은 JSON body에 정책값이 들어가는지 확인한다. +- caller response body는 provider fixture body와 byte-identical해야 한다. + +중간 검증: + +```bash +go test -count=1 ./apps/edge/internal/openai +``` + +## 최종 검증 + +```bash +go test -count=1 ./apps/edge/internal/openai +go test -count=1 ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai +``` + +기대 결과: 모두 PASS. 모든 코드 변경 완료 후 반드시 `CODE_REVIEW-local-G04.md`의 구현 에이전트 소유 섹션을 채운다. diff --git a/agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/CODE_REVIEW-cloud-G07.md b/agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/CODE_REVIEW-cloud-G07.md deleted file mode 100644 index 57188a3..0000000 --- a/agent-task/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/CODE_REVIEW-cloud-G07.md +++ /dev/null @@ -1,104 +0,0 @@ - - -# 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 selected Milestone `구현 잠금 > 결정 필요` item, fill `사용자 리뷰 요청` with linked evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. -> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation. -> Finalization is review-agent-only. - -## 개요 - -date=2026-07-07 -task=m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream, plan=0, tag=API - -## Roadmap Targets - -- Milestone: `agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md` -- Milestone link: [Milestone 문서](agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) -- Task ids: - - `edge-stream`: Edge OpenAI-compatible handler는 tunnel status/headers/body chunks를 caller에게 그대로 write하고 pure passthrough를 기본값으로 적용한다. -- Completion mode: check-on-pass - -## 이 파일을 읽는 리뷰 에이전트에게 - -> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다. - -각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요. - -## 구현 항목별 완료 여부 - -| 항목 | 완료 여부 | -|------|---------| -| [API-1] Tunnel Frame Routing | [ ] | -| [API-2] OpenAI Passthrough Writer | [ ] | - -## 구현 체크리스트 - -- [ ] `01_node_relay` predecessor `complete.log`를 확인하고 경로를 review stub에 기록한다. -- [ ] Edge transport가 `ProviderTunnelFrame`을 request-bound channel로 라우팅하되 `events.Bus`에는 publish하지 않게 구현한다. -- [ ] Edge service에 provider-pool/direct route 공통 `SubmitProviderTunnel` 경로를 추가하고 admission slot release를 END/ERROR/cancel에서 보장한다. -- [ ] OpenAI Chat Completions handler가 OpenAI-compatible provider route의 omitted mode를 pure `passthrough`로 처리하고 provider status/header/body bytes를 caller에게 그대로 쓴다. 검증: streaming/non-streaming Chat Completions fixture가 provider 원본과 byte-identical하게 반환된다. -- [ ] caller disconnect/context cancel이 Node cancel/control path로 전파되는지 test한다. -- [ ] `go test ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai`를 실행한다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. - -## 코드리뷰 전용 체크리스트 - -> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. - -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] predecessor `01_node_relay` completion evidence가 유효하다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G07_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_cloud_G07_M.log`로 아카이브한다. -- [ ] PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-openai-compatible-raw-tunnel-sideband-passthrough/02+01_edge_stream/`로 이동한다. - -## 계획 대비 변경 사항 - -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ - -## 주요 설계 결정 - -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ - -## 사용자 리뷰 요청 - -_기본값은 `없음`이다. 구현 중 새 결정이 필요해 보여도 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 이 섹션은 선택된 Milestone `구현 잠금 > 결정 필요` 항목이 실구현을 차단할 때만 채운다. 외부 환경/secret/서비스 준비, 검증 증거 공백, 반복 실패, 일반 범위 조정은 사용자 리뷰 요청이 아니며 `검증 결과`, `계획 대비 변경 사항`, 또는 code-review의 일반 follow-up plan으로 처리한다._ - -- 상태: 없음 -- 사유 유형: 없음 -- 연결 대상: 없음 -- 결정 필요: 없음 -- 차단 근거: 없음 -- 실행한 검증/명령: 없음 -- 자동 후속 불가 이유: 없음 -- 재개 조건: 없음 - -## 리뷰어를 위한 체크포인트 - -- Pure passthrough body에 IOP sideband/event/field가 섞이지 않는가. -- Streaming/non-streaming body byte comparison이 provider fixture와 동일한가. -- Provider-pool admission slot이 END/ERROR/cancel에서 release되는가. - -## 검증 결과 - -### API-1 중간 검증 -```bash -$ go test ./apps/edge/internal/transport ./apps/edge/internal/service -(output) -``` - -### API-2 중간 검증 -```bash -$ go test ./apps/edge/internal/openai -(output) -``` - -### 최종 검증 -```bash -$ go test ./apps/edge/internal/transport ./apps/edge/internal/service ./apps/edge/internal/openai -(output) -``` diff --git a/apps/edge/internal/bootstrap/runtime.go b/apps/edge/internal/bootstrap/runtime.go index 77e9323..832b673 100644 --- a/apps/edge/internal/bootstrap/runtime.go +++ b/apps/edge/internal/bootstrap/runtime.go @@ -96,6 +96,9 @@ func NewRuntime(cfg *config.EdgeConfig) (*Runtime, error) { func (r *Runtime) wireHandlers() { r.Server.SetRunEventHandler(r.EventBus.PublishRun) r.Server.SetNodeEventHandler(r.EventBus.PublishNode) + // Tunnel frames bypass the event bus: raw provider bytes go to the + // request-bound tunnel stream owned by the service. + r.Server.SetTunnelFrameHandler(r.Service.RouteProviderTunnelFrame) } func (r *Runtime) Start(ctx context.Context) error { diff --git a/apps/edge/internal/openai/chat_handler.go b/apps/edge/internal/openai/chat_handler.go index ee3ffaa..2011dbd 100644 --- a/apps/edge/internal/openai/chat_handler.go +++ b/apps/edge/internal/openai/chat_handler.go @@ -1,10 +1,12 @@ package openai import ( + "bytes" "context" "encoding/json" "errors" "fmt" + "io" "net/http" "path/filepath" "strconv" @@ -24,8 +26,16 @@ func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) { } defer r.Body.Close() + // The raw body is kept for the provider tunnel passthrough path, which + // forwards the caller's own payload (model rewritten to the served target). + rawBody, err := io.ReadAll(r.Body) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_request_error", "failed to read request body") + return + } + var req chatCompletionRequest - if err := decodeChatCompletionRequest(json.NewDecoder(r.Body), &req); err != nil { + if err := decodeChatCompletionRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil { writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error()) return } @@ -53,6 +63,16 @@ func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) { if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil { applyModelCatalogGenerationPolicyToChat(&req, *catalogEntry, outputPolicy.Strict) } + + // OpenAI-compatible provider routes default to pure passthrough (SDD D02): + // provider status/headers/body bytes are relayed to the caller unmodified + // over the raw tunnel instead of the normalized RunEvent path. + if passthroughResponseMode(runMeta) && routeUsesProviderTunnel(dispatch) { + estimate := s.estimateChatInputTokens(basePrompt, runMeta, req.Tools, req.ToolChoice) + contextClass := classifyContext(estimate, s.longContextThreshold()) + s.tunnelChatCompletionPassthrough(w, r, req, dispatch, runMeta, rawBody, estimate, contextClass) + return + } messages := req.Messages if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" { messages = prependSystemMessage(messages, instruction) @@ -248,6 +268,79 @@ func chatRunMetadata(runMeta map[string]string, req chatCompletionRequest, outpu return runMeta } +const ( + responseModeMetadataKey = "iop_response_mode" + responseModePassthrough = "passthrough" +) + +// passthroughResponseMode reports whether the request should use the pure +// passthrough tunnel: mode omitted (default, SDD D02) or explicitly +// "passthrough". Other explicit modes stay on the legacy normalized path +// until the mode-contract task lands their semantics. +func passthroughResponseMode(runMeta map[string]string) bool { + mode := strings.TrimSpace(runMeta[responseModeMetadataKey]) + return mode == "" || mode == responseModePassthrough +} + +// routeUsesProviderTunnel reports whether the resolved dispatch targets an +// OpenAI-compatible provider that serves raw tunnel passthrough. Provider-pool +// catalog routes and openai_compat/vllm type routes qualify; CLI and other +// legacy adapters keep the normalized RunEvent path. +func routeUsesProviderTunnel(d routeDispatch) bool { + if d.ProviderPool { + return true + } + switch strings.TrimSpace(d.Adapter) { + case "openai_compat", "vllm": + return true + default: + return false + } +} + +// rewriteChatCompletionModel replaces only the model field of the caller's +// original request JSON so the provider receives its served model name. The +// rest of the caller payload is forwarded without IOP rewriting. +func rewriteChatCompletionModel(rawBody []byte, target string, req chatCompletionRequest) ([]byte, error) { + if strings.TrimSpace(target) == "" && req.MaxTokens == nil && req.ThinkingTokenBudget == nil && req.Think == nil { + return rawBody, nil + } + var raw map[string]json.RawMessage + if err := json.Unmarshal(rawBody, &raw); err != nil { + return nil, fmt.Errorf("invalid JSON request") + } + if strings.TrimSpace(target) != "" { + modelJSON, err := json.Marshal(target) + if err != nil { + return nil, err + } + raw["model"] = modelJSON + } + if req.MaxTokens != nil { + maxTokensJSON, err := json.Marshal(*req.MaxTokens) + if err != nil { + return nil, err + } + raw["max_tokens"] = maxTokensJSON + delete(raw, "max_completion_tokens") + } + if req.ThinkingTokenBudget != nil { + budgetJSON, err := json.Marshal(*req.ThinkingTokenBudget) + if err != nil { + return nil, err + } + raw["thinking_token_budget"] = budgetJSON + } + if req.Think != nil { + thinkJSON, err := json.Marshal(*req.Think) + if err != nil { + return nil, err + } + raw["think"] = thinkJSON + } + return json.Marshal(raw) +} + func routeSupportsNativeToolCalls(dispatch routeDispatch) bool { if dispatch.ProviderPool { return true diff --git a/apps/edge/internal/openai/server.go b/apps/edge/internal/openai/server.go index 3d42445..6c1e903 100644 --- a/apps/edge/internal/openai/server.go +++ b/apps/edge/internal/openai/server.go @@ -17,6 +17,7 @@ import ( type runService interface { SubmitRun(context.Context, edgeservice.SubmitRunRequest) (edgeservice.RunResult, error) + SubmitProviderTunnel(context.Context, edgeservice.SubmitProviderTunnelRequest) (edgeservice.ProviderTunnelResult, error) OllamaAPI(context.Context, edgeservice.OllamaAPIRequest) (edgeservice.OllamaAPIView, error) CancelRun(context.Context, edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) } @@ -28,6 +29,15 @@ func (s *Server) cancelRunOnHTTPGiveUp(dispatch edgeservice.RunDispatch, err err if !isCancelWorthyRunError(err) || dispatch.RunID == "" { return } + s.sendCancelRun(dispatch) +} + +// sendCancelRun propagates cancellation for a dispatched run/tunnel to the +// Node cancel path. +func (s *Server) sendCancelRun(dispatch edgeservice.RunDispatch) { + if dispatch.RunID == "" { + return + } if _, cancelErr := s.service.CancelRun(context.Background(), edgeservice.CancelRunRequest{ NodeRef: dispatch.NodeID, RunID: dispatch.RunID, diff --git a/apps/edge/internal/openai/server_test.go b/apps/edge/internal/openai/server_test.go index a007a2d..c09a9aa 100644 --- a/apps/edge/internal/openai/server_test.go +++ b/apps/edge/internal/openai/server_test.go @@ -1,9 +1,11 @@ package openai import ( + "bytes" "context" "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" "strings" @@ -29,6 +31,151 @@ type fakeRunService struct { cancelMu sync.Mutex cancelCalls []edgeservice.CancelRunRequest cancelErr error + + // tunnelProviderURL, when set, makes SubmitProviderTunnel relay the tunnel + // request to this httptest provider fixture and stream the raw provider + // response back as ordered frames (emulating the Node relay). + tunnelProviderURL string + // tunnelServedTarget emulates provider-pool admission: BuildBody is called + // with this target when non-empty. + tunnelServedTarget string + // tunnelFrames, when set, is handed to the tunnel handle as-is instead of + // contacting tunnelProviderURL. + tunnelFrames chan *iop.ProviderTunnelFrame + tunnelErr error + tunnelReqs []edgeservice.SubmitProviderTunnelRequest + tunnelBodies [][]byte +} + +type fakeTunnelHandle struct { + dispatch edgeservice.RunDispatch + frames chan *iop.ProviderTunnelFrame +} + +func (h *fakeTunnelHandle) Dispatch() edgeservice.RunDispatch { return h.dispatch } +func (h *fakeTunnelHandle) Stream() edgeservice.ProviderTunnelStream { + return edgeservice.ProviderTunnelStream{Frames: h.frames} +} +func (h *fakeTunnelHandle) Close() {} +func (h *fakeTunnelHandle) WaitTimeout() time.Duration { return 5 * time.Second } + +func (s *fakeRunService) SubmitProviderTunnel(_ context.Context, req edgeservice.SubmitProviderTunnelRequest) (edgeservice.ProviderTunnelResult, error) { + s.submitMu.Lock() + s.tunnelReqs = append(s.tunnelReqs, req) + s.submitMu.Unlock() + if s.tunnelErr != nil { + return nil, s.tunnelErr + } + target := req.Target + if s.tunnelServedTarget != "" { + target = s.tunnelServedTarget + } + body := req.Body + if req.BuildBody != nil { + built, err := req.BuildBody(target) + if err != nil { + return nil, err + } + body = built + } + s.submitMu.Lock() + s.tunnelBodies = append(s.tunnelBodies, body) + s.submitMu.Unlock() + + frames := s.tunnelFrames + if frames == nil && s.tunnelProviderURL != "" { + relayed, err := relayProviderFixtureFrames(s.tunnelProviderURL, req.Method, req.Path, body) + if err != nil { + return nil, err + } + frames = relayed + } + return &fakeTunnelHandle{ + dispatch: edgeservice.RunDispatch{ + RunID: "run-tunnel", + NodeID: "node-1", + ModelGroupKey: req.ModelGroupKey, + Adapter: req.Adapter, + Target: target, + SessionID: req.SessionID, + TimeoutSec: 5, + }, + frames: frames, + }, nil +} + +func (s *fakeRunService) tunnelReqsSnapshot() []edgeservice.SubmitProviderTunnelRequest { + s.submitMu.Lock() + defer s.submitMu.Unlock() + return append([]edgeservice.SubmitProviderTunnelRequest(nil), s.tunnelReqs...) +} + +func (s *fakeRunService) tunnelBodiesSnapshot() [][]byte { + s.submitMu.Lock() + defer s.submitMu.Unlock() + return append([][]byte(nil), s.tunnelBodies...) +} + +// relayProviderFixtureFrames performs the provider HTTP request the way the +// Node relay does and converts status/header/body bytes into ordered tunnel +// frames, chunking the body at odd boundaries to exercise ordered writes. +func relayProviderFixtureFrames(providerURL, method, path string, body []byte) (chan *iop.ProviderTunnelFrame, error) { + httpReq, err := http.NewRequest(method, providerURL+path, bytes.NewReader(body)) + if err != nil { + return nil, err + } + httpReq.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(httpReq) + if err != nil { + return nil, err + } + defer resp.Body.Close() + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + headers := make(map[string]string) + for k, values := range resp.Header { + if len(values) > 0 { + headers[k] = values[0] + } + } + + const chunkSize = 7 + frames := make(chan *iop.ProviderTunnelFrame, len(respBody)/chunkSize+3) + seq := int64(0) + frames <- &iop.ProviderTunnelFrame{ + RunId: "run-tunnel", + TunnelId: "tunnel-1", + Sequence: seq, + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, + StatusCode: int32(resp.StatusCode), + Headers: headers, + } + seq++ + for off := 0; off < len(respBody); off += chunkSize { + end := off + chunkSize + if end > len(respBody) { + end = len(respBody) + } + frames <- &iop.ProviderTunnelFrame{ + RunId: "run-tunnel", + TunnelId: "tunnel-1", + Sequence: seq, + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, + Body: append([]byte(nil), respBody[off:end]...), + } + seq++ + } + frames <- &iop.ProviderTunnelFrame{ + RunId: "run-tunnel", + TunnelId: "tunnel-1", + Sequence: seq, + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, + End: true, + } + close(frames) + return frames, nil } func (s *fakeRunService) CancelRun(_ context.Context, req edgeservice.CancelRunRequest) (edgeservice.CommandResult, error) { @@ -221,6 +368,7 @@ func TestChatCompletionsPreservesProviderFinishReason(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"hi"}] }`)) @@ -317,6 +465,7 @@ func TestChatCompletionsOmitsProviderToolChoiceAuto(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"hi"}], "tools":[{"type":"function","function":{"name":"lookup"}}], @@ -349,6 +498,7 @@ func TestChatCompletionsPassesProviderForcedToolChoice(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"hi"}], "tools":[{"type":"function","function":{"name":"lookup"}}], @@ -517,6 +667,7 @@ func TestChatCompletionsSynthesizesTextToolCallsForProviderRoute(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object"}}}], @@ -549,6 +700,7 @@ func TestChatCompletionsSynthesizesEditToolCallWithMarkdownAndAngleBracketPlaceh srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"edit"}], "tools":[{"type":"function","function":{"name":"edit","parameters":{"type":"object","properties":{"path":{"type":"string"},"edits":{"type":"array","items":{"type":"object","properties":{"oldText":{"type":"string"},"newText":{"type":"string"}},"required":["newText"],"additionalProperties":false}}},"required":["path","edits"],"additionalProperties":false}}}], @@ -599,6 +751,7 @@ func TestChatCompletionsStreamSynthesizesEditToolCallWithMarkdownAndAngleBracket srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "stream":true, "messages":[{"role":"user","content":"edit"}], @@ -662,6 +815,7 @@ func TestChatCompletionsSynthesizesProviderTextToolCallsWhenFallbackMarked(t *te srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}},"required":["commands"]}}}], @@ -707,6 +861,7 @@ func TestChatCompletionsSanitizesKnownSentinelToken(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"hi"}] }`)) @@ -774,6 +929,7 @@ func TestChatCompletionsPassesThroughNativeToolCallsFromProviderRoute(t *testing srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands"}}], @@ -816,6 +972,7 @@ func TestChatCompletionsNativeToolCallPreservesLeadingProse(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands"}}], @@ -861,6 +1018,7 @@ func TestChatCompletionsNormalizesNativeToolCallArgumentsForProviderRoute(t *tes srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}},"required":["commands"]}}}], @@ -911,6 +1069,7 @@ func TestChatCompletionsRetriesMalformedToolCallBeforeResponse(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}},"required":["commands"],"additionalProperties":false}}}], @@ -970,6 +1129,7 @@ func TestChatCompletionsFailsMalformedToolCallAfterRetryLimit(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}},"required":["commands"],"additionalProperties":false}}}], @@ -1002,6 +1162,7 @@ func TestChatCompletionsToolChoiceNoneSkipsRuntimeToolValidation(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object","properties":{"commands":{"type":"array","items":{"type":"string"}}},"required":["commands"]}}}], @@ -1046,6 +1207,7 @@ func TestChatCompletionsToolsStreamRetriesMalformedToolCallBeforeChunk(t *testin srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "stream":true, "messages":[{"role":"user","content":"edit"}], @@ -1104,6 +1266,7 @@ func TestChatCompletionsStrictBufferedStreamRetriesMalformedToolCallBeforeChunk( srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat", StrictOutput: true, StrictStreamBuffer: true}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -1156,6 +1319,7 @@ func TestChatCompletionsStrictBufferedStreamFailsMalformedToolCallAfterRetryLimi srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat", StrictOutput: true, StrictStreamBuffer: true}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -1659,6 +1823,7 @@ func TestChatCompletionsMapsMaxCompletionTokens(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"from-request", "messages":[{"role":"user","content":"hi"}], "max_completion_tokens":7 @@ -1748,6 +1913,7 @@ func TestChatCompletionsStreamWithToolsBuffersTextUntilComplete(t *testing.T) { )} srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"explain sudo"}], @@ -1831,6 +1997,7 @@ func TestChatCompletionsStreamPassesThroughNativeToolCallsFromProviderRoute(t *t srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -1866,6 +2033,7 @@ func TestChatCompletionsStreamDropsRawTextWhenNativeToolCallsArrive(t *testing.T srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -1901,6 +2069,7 @@ func TestChatCompletionsToolsStreamNativeToolCallPreservesLeadingProse(t *testin srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -1960,6 +2129,7 @@ func TestChatCompletionsStreamDropsPartialTextToolCallSuffixWhenNativeToolCallsA srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"status"}], @@ -2246,6 +2416,7 @@ func TestChatCompletionsStreamSanitizesKnownSentinelTokenAcrossDeltas(t *testing srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"stream-model", "stream":true, "messages":[{"role":"user","content":"hi"}] @@ -3434,6 +3605,7 @@ func TestChatCompletionsRouteCatalogDispatchesModelB(t *testing.T) { }, }, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"model-b", "messages":[{"role":"user","content":"hi"}] }`)) @@ -3849,12 +4021,16 @@ func TestHandleModelsProviderPoolCatalog(t *testing.T) { } // TestChatCompletionsProviderPoolDispatch verifies that when a request model -// matches the provider-pool catalog, ProviderPool=true is set on the service -// request and Adapter/Target are left empty for service-layer resolution. +// matches the provider-pool catalog with omitted response mode, the request is +// dispatched over the provider tunnel with ProviderPool=true and Adapter/Target +// left empty for service-layer resolution (pure passthrough default, SDD D02). func TestChatCompletionsProviderPoolDispatch(t *testing.T) { - fake := &fakeRunService{events: make(chan *iop.RunEvent, 2)} - fake.events <- &iop.RunEvent{Type: "delta", Delta: "ok"} - fake.events <- &iop.RunEvent{Type: "complete"} + frames := make(chan *iop.ProviderTunnelFrame, 3) + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200} + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte(`{"ok":true}`)} + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true} + close(frames) + fake := &fakeRunService{tunnelFrames: frames} catalog := []config.ModelCatalogEntry{ {ID: "qwen3.6:35b", Providers: map[string]string{"prov-vllm": "Qwen3-35B-A22B"}}, @@ -3872,14 +4048,386 @@ func TestChatCompletionsProviderPoolDispatch(t *testing.T) { if w.Code != http.StatusOK { t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) } - if !fake.req.ProviderPool { + tunnelReqs := fake.tunnelReqsSnapshot() + if len(tunnelReqs) != 1 { + t.Fatalf("expected 1 tunnel dispatch, got %d (SubmitRun calls: %d)", len(tunnelReqs), len(fake.reqsSnapshot())) + } + treq := tunnelReqs[0] + if !treq.ProviderPool { t.Error("ProviderPool should be true for catalog-matched model") } - if fake.req.ModelGroupKey != "qwen3.6:35b" { - t.Errorf("ModelGroupKey: got %q, want qwen3.6:35b", fake.req.ModelGroupKey) + if treq.ModelGroupKey != "qwen3.6:35b" { + t.Errorf("ModelGroupKey: got %q, want qwen3.6:35b", treq.ModelGroupKey) } - if fake.req.Adapter != "" || fake.req.Target != "" { - t.Errorf("Adapter/Target should be empty for provider-pool dispatch, got %q/%q", fake.req.Adapter, fake.req.Target) + if treq.Adapter != "" || treq.Target != "" { + t.Errorf("Adapter/Target should be empty for provider-pool dispatch, got %q/%q", treq.Adapter, treq.Target) + } + if treq.Method != http.MethodPost || treq.Path != "/v1/chat/completions" { + t.Errorf("tunnel method/path: got %q %q", treq.Method, treq.Path) + } + if len(fake.reqsSnapshot()) != 0 { + t.Errorf("normalized SubmitRun must not be called for omitted-mode provider route") + } +} + +// chatPassthroughServer returns an openai.Server whose fake service relays the +// tunnel to the given provider fixture, emulating the Node relay. +func chatPassthroughServer(providerURL string) (*Server, *fakeRunService) { + fake := &fakeRunService{ + tunnelProviderURL: providerURL, + tunnelServedTarget: "served-model", + } + srv := NewServer(config.EdgeOpenAIConf{}, fake, nil) + srv.SetModelCatalog([]config.ModelCatalogEntry{ + {ID: "pool-model", Providers: map[string]string{"prov-1": "served-model"}}, + }) + return srv, fake +} + +// TestChatCompletionsPassthroughNonStreamingByteIdentity verifies SDD S04 for +// the non-streaming path: with omitted response mode, the caller receives the +// provider JSON body byte-identically, along with provider status and headers. +func TestChatCompletionsPassthroughNonStreamingByteIdentity(t *testing.T) { + // Provider-original body with fields the normalized path would rewrite or + // drop (reasoning_content, provider-specific key order and spacing). + providerBody := "{\"id\":\"cmpl-provider-1\",\"object\":\"chat.completion\", \"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"hi\",\"reasoning_content\":\"because\"},\"finish_reason\":\"stop\"}],\"provider_extra\":{\"a\":1}}" + var gotProviderReq []byte + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotProviderReq, _ = io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Provider-Marker", "prov-42") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(providerBody)) + })) + defer provider.Close() + + srv, fake := chatPassthroughServer(provider.URL) + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if got := w.Body.String(); got != providerBody { + t.Fatalf("body not byte-identical to provider:\n got: %q\nwant: %q", got, providerBody) + } + if got := w.Header().Get("X-Provider-Marker"); got != "prov-42" { + t.Errorf("provider header not relayed: %q", got) + } + if got := w.Header().Get("Content-Type"); got != "application/json" { + t.Errorf("content type not relayed: %q", got) + } + // The provider request body carries the served model (model rewrite only). + if !strings.Contains(string(gotProviderReq), `"model":"served-model"`) { + t.Errorf("provider request body missing served model: %s", gotProviderReq) + } + if !strings.Contains(string(gotProviderReq), `"content":"hello"`) { + t.Errorf("provider request body lost caller messages: %s", gotProviderReq) + } + if len(fake.reqsSnapshot()) != 0 { + t.Error("normalized SubmitRun must not be called on passthrough") + } +} + +// TestChatCompletionsPassthroughProviderPoolGenerationPolicy verifies that +// provider-pool omitted-mode passthrough requests apply the catalog entry's +// generation policy (default_max_tokens, min_max_tokens, default_thinking_token_budget) +// before dispatching to the provider, while response bytes remain byte-identical. +func TestChatCompletionsPassthroughProviderPoolGenerationPolicy(t *testing.T) { + providerBody := `{"id":"cmpl-provider-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}` + var gotProviderReq []byte + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotProviderReq, _ = io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(providerBody)) + })) + defer provider.Close() + + fake := &fakeRunService{ + tunnelProviderURL: provider.URL, + tunnelServedTarget: "served-model", + } + // Enable strict output to trigger think=true from policy + srv := NewServer(config.EdgeOpenAIConf{StrictOutput: true}, fake, nil) + srv.SetModelCatalog([]config.ModelCatalogEntry{ + { + ID: "pool-model", + Providers: map[string]string{"prov-1": "served-model"}, + DefaultMaxTokens: 100, + MinMaxTokens: 50, + DefaultThinkingTokenBudget: 30, + }, + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}], + "max_completion_tokens":20 + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if got := w.Body.String(); got != providerBody { + t.Fatalf("body not byte-identical to provider:\n got: %q\nwant: %q", got, providerBody) + } + + var parsedReq map[string]interface{} + if err := json.Unmarshal(gotProviderReq, &parsedReq); err != nil { + t.Fatalf("failed to parse provider request: %v", err) + } + + if model, ok := parsedReq["model"].(string); !ok || model != "served-model" { + t.Errorf("expected model served-model, got %v", parsedReq["model"]) + } + + // MinMaxTokens (50) should reach max_tokens (overriding max_completion_tokens 20) + if maxTokens, ok := parsedReq["max_tokens"].(float64); !ok || maxTokens != 50 { + t.Errorf("expected max_tokens 50, got %v", parsedReq["max_tokens"]) + } + if _, ok := parsedReq["max_completion_tokens"]; ok { + t.Errorf("max_completion_tokens should be deleted") + } + + // DefaultThinkingTokenBudget (30) and Think (true) should reach provider + if budget, ok := parsedReq["thinking_token_budget"].(float64); !ok || budget != 30 { + t.Errorf("expected thinking_token_budget 30, got %v", parsedReq["thinking_token_budget"]) + } + if think, ok := parsedReq["think"].(bool); !ok || !think { + t.Errorf("expected think true, got %v", parsedReq["think"]) + } +} + +// TestChatCompletionsPassthroughStreamingByteIdentity verifies SDD S04 for the +// streaming path: provider SSE bytes (including provider-specific fields like +// reasoning_content and native tool_calls chunks) reach the caller +// byte-identically with no IOP rewriting. +func TestChatCompletionsPassthroughStreamingByteIdentity(t *testing.T) { + providerBody := "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"thinking...\"}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{\"content\":\"not rewritten\"}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"name\":\"f\",\"arguments\":\"{}\"}}]}}]}\n\n" + + "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":3}}\n\n" + + "data: [DONE]\n\n" + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(providerBody)) + })) + defer provider.Close() + + srv, _ := chatPassthroughServer(provider.URL) + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "stream":true, + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if got := w.Body.String(); got != providerBody { + t.Fatalf("SSE stream not byte-identical to provider:\n got: %q\nwant: %q", got, providerBody) + } + if got := w.Header().Get("Content-Type"); got != "text/event-stream" { + t.Errorf("content type not relayed: %q", got) + } +} + +// TestChatCompletionsPassthroughProviderErrorStatusRelayed verifies a provider +// error status/body is passed to the caller unmodified instead of being +// converted into an IOP error envelope. +func TestChatCompletionsPassthroughProviderErrorStatusRelayed(t *testing.T) { + providerBody := `{"error":{"message":"rate limited by provider","type":"rate_limit"}}` + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(providerBody)) + })) + defer provider.Close() + + srv, _ := chatPassthroughServer(provider.URL) + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusTooManyRequests { + t.Fatalf("provider status not relayed: got %d", w.Code) + } + if got := w.Body.String(); got != providerBody { + t.Fatalf("provider error body not byte-identical: %q", got) + } +} + +// TestChatCompletionsPassthroughLegacyProviderRouteUsesTunnel verifies that a +// legacy model_routes entry with an openai_compat adapter also defaults to the +// tunnel passthrough path (direct route, no provider pool). +func TestChatCompletionsPassthroughLegacyProviderRouteUsesTunnel(t *testing.T) { + frames := make(chan *iop.ProviderTunnelFrame, 3) + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200} + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte(`{"ok":true}`)} + frames <- &iop.ProviderTunnelFrame{Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true} + close(frames) + fake := &fakeRunService{tunnelFrames: frames} + srv := NewServer(config.EdgeOpenAIConf{ + ModelRoutes: []config.OpenAIRouteEntry{ + {Model: "compat-model", Adapter: "openai_compat", Target: "backend-model"}, + }, + }, fake, nil) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"compat-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + tunnelReqs := fake.tunnelReqsSnapshot() + if len(tunnelReqs) != 1 { + t.Fatalf("expected tunnel dispatch for openai_compat route, got %d", len(tunnelReqs)) + } + if tunnelReqs[0].Adapter != "openai_compat" || tunnelReqs[0].Target != "backend-model" { + t.Errorf("tunnel adapter/target: got %q/%q", tunnelReqs[0].Adapter, tunnelReqs[0].Target) + } + if tunnelReqs[0].ProviderPool { + t.Error("legacy route must not set ProviderPool") + } + bodies := fake.tunnelBodiesSnapshot() + if len(bodies) != 1 || !strings.Contains(string(bodies[0]), `"model":"backend-model"`) { + t.Errorf("legacy route body must carry route target: %s", bodies) + } +} + +// TestChatCompletionsExplicitNonPassthroughModeKeepsNormalizedPath verifies +// that an explicit non-passthrough response mode stays on the legacy +// normalized RunEvent path (mode semantics land with the mode-contract task). +func TestChatCompletionsExplicitNonPassthroughModeKeepsNormalizedPath(t *testing.T) { + fake := &fakeRunService{events: bufferedRunEvents( + &iop.RunEvent{Type: "delta", Delta: "ok"}, + &iop.RunEvent{Type: "complete"}, + )} + srv := NewServer(config.EdgeOpenAIConf{}, fake, nil) + srv.SetModelCatalog([]config.ModelCatalogEntry{ + {ID: "pool-model", Providers: map[string]string{"prov-1": "served-model"}}, + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}], + "metadata":{"iop_response_mode":"transformed"} + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if len(fake.tunnelReqsSnapshot()) != 0 { + t.Error("explicit non-passthrough mode must not use the tunnel") + } + if len(fake.reqsSnapshot()) != 1 { + t.Fatalf("expected normalized SubmitRun dispatch, got %d", len(fake.reqsSnapshot())) + } +} + +// TestChatCompletionsPassthroughNonProviderRouteKeepsNormalizedPath verifies +// that CLI/ollama legacy routes are unaffected by the passthrough default. +func TestChatCompletionsPassthroughNonProviderRouteKeepsNormalizedPath(t *testing.T) { + fake := &fakeRunService{events: bufferedRunEvents( + &iop.RunEvent{Type: "delta", Delta: "ok"}, + &iop.RunEvent{Type: "complete"}, + )} + srv := NewServer(config.EdgeOpenAIConf{Adapter: "ollama", Target: "llama-fixed"}, fake, nil) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"any-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if len(fake.tunnelReqsSnapshot()) != 0 { + t.Error("ollama route must not use the tunnel") + } + if len(fake.reqsSnapshot()) != 1 { + t.Fatalf("expected normalized SubmitRun dispatch, got %d", len(fake.reqsSnapshot())) + } +} + +// TestChatCompletionsPassthroughTunnelErrorBeforeStartReturns502 verifies an +// ERROR frame before response-start becomes a JSON error response. +func TestChatCompletionsPassthroughTunnelErrorBeforeStartReturns502(t *testing.T) { + frames := make(chan *iop.ProviderTunnelFrame, 1) + frames <- &iop.ProviderTunnelFrame{ + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, + Error: "provider connect refused", + } + close(frames) + fake := &fakeRunService{tunnelFrames: frames} + srv := NewServer(config.EdgeOpenAIConf{}, fake, nil) + srv.SetModelCatalog([]config.ModelCatalogEntry{ + {ID: "pool-model", Providers: map[string]string{"prov-1": "served-model"}}, + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusBadGateway { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "provider connect refused") { + t.Fatalf("error message not surfaced: %s", w.Body.String()) + } +} + +// TestChatCompletionsPassthroughContextCancelSendsCancelRun verifies caller +// disconnect propagates to the Node cancel path for tunnel dispatches. +func TestChatCompletionsPassthroughContextCancelSendsCancelRun(t *testing.T) { + // Frames channel stays open and empty: the tunnel is in-flight when the + // caller context is cancelled. + fake := &fakeRunService{tunnelFrames: make(chan *iop.ProviderTunnelFrame)} + srv := NewServer(config.EdgeOpenAIConf{}, fake, nil) + srv.SetModelCatalog([]config.ModelCatalogEntry{ + {ID: "pool-model", Providers: map[string]string{"prov-1": "served-model"}}, + }) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"pool-model", + "messages":[{"role":"user","content":"hello"}] + }`)).WithContext(ctx) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + calls := fake.cancelCallsSnapshot() + if len(calls) != 1 { + t.Fatalf("expected 1 CancelRun call, got %d: %+v", len(calls), calls) + } + if calls[0].RunID != "run-tunnel" || calls[0].NodeRef != "node-1" { + t.Fatalf("unexpected cancel request: %+v", calls[0]) } } @@ -3898,10 +4446,14 @@ func TestChatCompletionsProviderPoolAppliesGenerationPolicy(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{}, fake, nil) srv.SetModelCatalog(catalog) + // Explicit non-passthrough mode keeps the normalized RunEvent path where + // catalog generation policy applies; omitted mode now defaults to tunnel + // passthrough for provider-pool routes. req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ "model":"ornith:35b", "messages":[{"role":"user","content":"hello"}], - "max_tokens":4096 + "max_tokens":4096, + "metadata":{"iop_response_mode":"transformed"} }`)) w := httptest.NewRecorder() srv.handleChatCompletions(w, req) @@ -3936,7 +4488,8 @@ func TestChatCompletionsProviderPoolThinkingPolicyOverridesStrictOutputDisable(t req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ "model":"ornith:35b", - "messages":[{"role":"user","content":"hello"}] + "messages":[{"role":"user","content":"hello"}], + "metadata":{"iop_response_mode":"transformed"} }`)) w := httptest.NewRecorder() srv.handleChatCompletions(w, req) @@ -3971,7 +4524,8 @@ func TestChatCompletionsProviderPoolPreservesLargerGenerationPolicyValues(t *tes "model":"ornith:35b", "messages":[{"role":"user","content":"hello"}], "max_tokens":40000, - "thinking_token_budget":2048 + "thinking_token_budget":2048, + "metadata":{"iop_response_mode":"transformed"} }`)) w := httptest.NewRecorder() srv.handleChatCompletions(w, req) @@ -4242,6 +4796,7 @@ func TestChatCompletionsProviderStreamSynthesizesTextToolCalls(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object"}}}], @@ -4281,6 +4836,7 @@ func TestChatCompletionsProviderStreamBlocksUnknownTextToolCall(t *testing.T) { srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands"}}], @@ -4352,6 +4908,7 @@ func TestChatCompletionsFailsMalformedTextToolCallAfterRetryLimit(t *testing.T) srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object"}}}], @@ -4390,6 +4947,7 @@ func TestChatCompletionsProviderStreamBlocksMalformedTextToolCall(t *testing.T) srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[{"type":"function","function":{"name":"run_commands","parameters":{"type":"object"}}}], @@ -4514,6 +5072,7 @@ func TestChatCompletionsFailsWhenAnySynthesizedTextToolCallViolatesSchema(t *tes srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "metadata":{"iop_response_mode":"transformed"}, "model":"qwen3.6:35b", "messages":[{"role":"user","content":"status"}], "tools":[ diff --git a/apps/edge/internal/openai/stream.go b/apps/edge/internal/openai/stream.go index 5c64dc9..e39347b 100644 --- a/apps/edge/internal/openai/stream.go +++ b/apps/edge/internal/openai/stream.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "os" + "strconv" "strings" "time" "unicode" @@ -12,6 +13,7 @@ import ( "go.uber.org/zap" edgeservice "iop/apps/edge/internal/service" + iop "iop/proto/gen/iop" ) const ( @@ -304,6 +306,193 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re } } +// tunnelChatCompletionPassthrough serves a Chat Completions request over the +// raw provider tunnel (SDD S04/S06): provider status, headers, and body bytes +// are written to the caller byte-identically. No IOP sideband fields or events +// are added to the response body. +func (s *Server) tunnelChatCompletionPassthrough(w http.ResponseWriter, r *http.Request, req chatCompletionRequest, dispatch routeDispatch, runMeta map[string]string, rawBody []byte, estimate int, contextClass string) { + metadata := make(map[string]string, len(runMeta)+5) + for k, v := range runMeta { + metadata[k] = v + } + metadata["openai_model"] = req.Model + metadata["openai_stream"] = strconv.FormatBool(req.Stream) + metadata[responseModeMetadataKey] = responseModePassthrough + metadata["estimated_input_tokens"] = strconv.Itoa(estimate) + metadata["context_class"] = contextClass + + tunnelReq := edgeservice.SubmitProviderTunnelRequest{ + NodeRef: dispatch.NodeRef, + ModelGroupKey: strings.TrimSpace(req.Model), + Adapter: dispatch.Adapter, + Target: dispatch.Target, + SessionID: dispatch.SessionID, + Method: http.MethodPost, + Path: "/v1/chat/completions", + BuildBody: func(target string) ([]byte, error) { + return rewriteChatCompletionModel(rawBody, target, req) + }, + Stream: req.Stream, + TimeoutSec: dispatch.TimeoutSec, + MaxQueue: dispatch.MaxQueue, + QueueTimeoutMS: dispatch.QueueTimeoutMS, + Metadata: metadata, + EstimatedInputTokens: estimate, + ContextClass: contextClass, + ProviderPool: dispatch.ProviderPool, + } + + handle, err := s.service.SubmitProviderTunnel(r.Context(), tunnelReq) + if err != nil { + writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error()) + return + } + defer handle.Close() + + s.logger.Info("openai chat completion passthrough dispatch", + zap.String("run_id", handle.Dispatch().RunID), + zap.String("node_id", handle.Dispatch().NodeID), + zap.String("model_group", handle.Dispatch().ModelGroupKey), + zap.String("adapter", handle.Dispatch().Adapter), + zap.String("target", handle.Dispatch().Target), + zap.Bool("stream", req.Stream), + zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens), + zap.String("context_class", handle.Dispatch().ContextClass), + zap.String("queue_reason", handle.Dispatch().QueueReason), + ) + + s.writeProviderTunnelResponse(w, r, handle) +} + +// writeProviderTunnelResponse relays ordered tunnel frames to the HTTP caller. +// The response-start frame sets status/headers, body frames are written and +// flushed in order, and END terminates the response. Caller disconnect and +// wait timeout propagate cancellation to the Node cancel path. +func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Request, handle edgeservice.ProviderTunnelResult) { + frames := handle.Stream().Frames + if frames == nil { + writeError(w, http.StatusBadGateway, "provider_tunnel_error", "tunnel stream unavailable") + return + } + flusher, _ := w.(http.Flusher) + timer := time.NewTimer(handle.WaitTimeout()) + defer timer.Stop() + + wroteHeader := false + bodyBytes := 0 + defer func() { + s.logger.Info("openai chat completion passthrough closed", + zap.String("run_id", handle.Dispatch().RunID), + zap.Bool("wrote_header", wroteHeader), + zap.Int("body_bytes", bodyBytes), + ) + }() + + for { + select { + case <-r.Context().Done(): + s.cancelRunOnHTTPGiveUp(handle.Dispatch(), r.Context().Err()) + return + case <-timer.C: + s.cancelRunOnHTTPGiveUp(handle.Dispatch(), errRunTimedOut) + if !wroteHeader { + writeError(w, http.StatusBadGateway, "run_error", errRunTimedOut.Error()) + } + return + case frame, ok := <-frames: + if !ok { + if !wroteHeader { + writeError(w, http.StatusBadGateway, "provider_tunnel_error", "tunnel stream closed before provider response") + } + return + } + switch frame.GetKind() { + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START: + if wroteHeader { + continue + } + copyProviderResponseHeaders(w.Header(), frame.GetHeaders()) + status := int(frame.GetStatusCode()) + if status == 0 { + status = http.StatusOK + } + w.WriteHeader(status) + wroteHeader = true + if flusher != nil { + flusher.Flush() + } + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY: + body := frame.GetBody() + if len(body) == 0 { + continue + } + if !wroteHeader { + // Defensive: a body frame before response-start still reaches the + // caller instead of being dropped. + w.WriteHeader(http.StatusOK) + wroteHeader = true + } + if _, err := w.Write(body); err != nil { + // The caller is gone mid-stream; propagate cancel to the Node. + s.sendCancelRun(handle.Dispatch()) + return + } + bodyBytes += len(body) + if flusher != nil { + flusher.Flush() + } + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR: + msg := frame.GetError() + if msg == "" { + msg = "provider tunnel failed" + } + if !wroteHeader { + writeError(w, http.StatusBadGateway, "provider_tunnel_error", msg) + return + } + // Status/headers are already committed; the response is truncated + // and the caller observes the broken stream. + s.logger.Warn("openai passthrough tunnel error after response start", + zap.String("run_id", handle.Dispatch().RunID), + zap.String("error", msg), + ) + return + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END: + if !wroteHeader { + writeError(w, http.StatusBadGateway, "provider_tunnel_error", "tunnel ended before provider response") + } + return + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_USAGE: + // Usage frames are sideband observation candidates; pure passthrough + // never merges them into the response body (SDD D04). + continue + } + } + } +} + +// hopByHopResponseHeaders are transport-level headers owned by each hop; they +// are not copied from the provider response to the caller response. +var hopByHopResponseHeaders = map[string]struct{}{ + "Connection": {}, + "Keep-Alive": {}, + "Proxy-Authenticate": {}, + "Proxy-Authorization": {}, + "Te": {}, + "Trailer": {}, + "Transfer-Encoding": {}, + "Upgrade": {}, +} + +func copyProviderResponseHeaders(dst http.Header, headers map[string]string) { + for k, v := range headers { + if _, hop := hopByHopResponseHeaders[http.CanonicalHeaderKey(k)]; hop { + continue + } + dst.Set(k, v) + } +} + type streamToolTextFilter struct { pending string } diff --git a/apps/edge/internal/service/run_dispatch.go b/apps/edge/internal/service/run_dispatch.go index e8d3f95..310577f 100644 --- a/apps/edge/internal/service/run_dispatch.go +++ b/apps/edge/internal/service/run_dispatch.go @@ -711,6 +711,352 @@ func (s *Service) dispatchToEntry(entry *edgenode.NodeEntry, req SubmitRunReques }, nil } +// tunnelFrameBuffer is the per-tunnel channel capacity. The router blocks on a +// full buffer instead of dropping so ordered passthrough bytes are never lost +// while the subscriber is alive. +const tunnelFrameBuffer = 4096 + +// providerTunnelRouter delivers ProviderTunnelFrame messages to the +// request-bound subscriber of their tunnel id. It is intentionally separate +// from events.Bus: tunnel body bytes are the passthrough source of truth and +// must not enter the lossy run event fanout. +type providerTunnelRouter struct { + mu sync.Mutex + subs map[string]*providerTunnelSub +} + +type providerTunnelSub struct { + ch chan *iop.ProviderTunnelFrame + done chan struct{} +} + +func newProviderTunnelRouter() *providerTunnelRouter { + return &providerTunnelRouter{subs: make(map[string]*providerTunnelSub)} +} + +func (r *providerTunnelRouter) subscribe(tunnelID string, buffer int) (<-chan *iop.ProviderTunnelFrame, func()) { + sub := &providerTunnelSub{ + ch: make(chan *iop.ProviderTunnelFrame, buffer), + done: make(chan struct{}), + } + r.mu.Lock() + r.subs[tunnelID] = sub + r.mu.Unlock() + var once sync.Once + unsubscribe := func() { + once.Do(func() { + r.mu.Lock() + if cur, ok := r.subs[tunnelID]; ok && cur == sub { + delete(r.subs, tunnelID) + } + r.mu.Unlock() + close(sub.done) + }) + } + return sub.ch, unsubscribe +} + +// route delivers a frame to its tunnel subscriber. Frames without a subscriber +// are dropped (the request already finished or was cancelled). +func (r *providerTunnelRouter) route(frame *iop.ProviderTunnelFrame) { + if frame == nil { + return + } + r.mu.Lock() + sub := r.subs[frame.GetTunnelId()] + r.mu.Unlock() + if sub == nil { + return + } + select { + case sub.ch <- frame: + case <-sub.done: + } +} + +// RouteProviderTunnelFrame delivers a ProviderTunnelFrame received from Edge +// transport to its request-bound tunnel stream. It never publishes to the +// run/node event bus. +func (s *Service) RouteProviderTunnelFrame(frame *iop.ProviderTunnelFrame) { + if s == nil || s.tunnels == nil { + return + } + s.tunnels.route(frame) +} + +// SubmitProviderTunnelRequest asks a node to open a raw provider HTTP request +// and relay the response as ordered ProviderTunnelFrame messages. It is the +// passthrough sibling of SubmitRunRequest and shares the provider-pool +// admission gate. +type SubmitProviderTunnelRequest struct { + NodeRef string + RunID string + ModelGroupKey string + Adapter string + Target string + SessionID string + Method string + Path string + Headers map[string]string + Body []byte + // BuildBody, when set, produces the provider request body from the final + // resolved target (provider-pool admission rewrites the target to the + // winning candidate's served model). It takes precedence over Body. + BuildBody func(target string) ([]byte, error) + Stream bool + TimeoutSec int + MaxQueue int + QueueTimeoutMS int + Metadata map[string]string + EstimatedInputTokens int + ContextClass string + ProviderPool bool +} + +// ProviderTunnelStream carries the ordered raw provider frames of a dispatched +// tunnel. The channel is closed after the terminal END/ERROR frame or Close. +type ProviderTunnelStream struct { + Frames <-chan *iop.ProviderTunnelFrame +} + +// ProviderTunnelResult is the surface-neutral handle for a dispatched provider +// tunnel, mirroring RunResult for the raw passthrough path. +type ProviderTunnelResult interface { + Dispatch() RunDispatch + Stream() ProviderTunnelStream + Close() + WaitTimeout() time.Duration +} + +// ProviderTunnelHandle implements ProviderTunnelResult for tunnels dispatched +// over the Edge-Node socket. +type ProviderTunnelHandle struct { + RunDispatch + TunnelID string + frames <-chan *iop.ProviderTunnelFrame + closeOnce sync.Once + close func() +} + +func (h *ProviderTunnelHandle) Dispatch() RunDispatch { + if h == nil { + return RunDispatch{} + } + return h.RunDispatch +} + +func (h *ProviderTunnelHandle) Stream() ProviderTunnelStream { + if h == nil { + return ProviderTunnelStream{} + } + return ProviderTunnelStream{Frames: h.frames} +} + +func (h *ProviderTunnelHandle) Close() { + if h != nil && h.close != nil { + h.closeOnce.Do(h.close) + } +} + +func (h *ProviderTunnelHandle) WaitTimeout() time.Duration { + if h == nil { + return time.Duration(DefaultTimeoutSec+5) * time.Second + } + return time.Duration(normalizeTimeoutSec(h.TimeoutSec)+5) * time.Second +} + +// SubmitProviderTunnel dispatches a raw provider tunnel request. Provider-pool +// requests go through the same admission gate as SubmitRun; the reserved slot +// is released when the tunnel reaches END/ERROR or the handle is closed +// (cancel), never via the run event bus. +func (s *Service) SubmitProviderTunnel(ctx context.Context, req SubmitProviderTunnelRequest) (ProviderTunnelResult, error) { + if req.ProviderPool && req.ModelGroupKey != "" && s.queue != nil { + return s.submitProviderTunnelQueued(ctx, req) + } + return s.submitProviderTunnelDirect(req) +} + +func (s *Service) submitProviderTunnelQueued(ctx context.Context, req SubmitProviderTunnelRequest) (ProviderTunnelResult, error) { + candidates, policy, err := s.resolveQueueCandidates(SubmitRunRequest{ + NodeRef: req.NodeRef, + ModelGroupKey: req.ModelGroupKey, + Adapter: req.Adapter, + Target: req.Target, + MaxQueue: req.MaxQueue, + QueueTimeoutMS: req.QueueTimeoutMS, + EstimatedInputTokens: req.EstimatedInputTokens, + ContextClass: req.ContextClass, + ProviderPool: true, + }) + if err != nil { + return nil, err + } + + long := req.ContextClass == contextClassLong + selected, queueReason, err := s.queue.admitWithReason(ctx, req.ModelGroupKey, req.Adapter, req.Target, candidates, policy, long) + if err != nil { + return nil, err + } + longReserved := long && selected.longContextCapacity > 0 + + adapter := req.Adapter + if selected.adapter != "" { + adapter = selected.adapter + } + target := req.Target + if selected.servedTarget != "" { + target = selected.servedTarget + } + + tunnelReq, runID, err := buildProviderTunnelRequest(req, adapter, target) + if err != nil { + s.queue.releaseSlotWithLong(req.ModelGroupKey, selected.entry.NodeID, selected.providerID, longReserved) + return nil, err + } + + // Track inflight before send so END/ERROR/close release paths can find it + // even if the terminal frame arrives before Send returns. + s.queue.trackInflight(req.ModelGroupKey, runID, selected.entry.NodeID, selected.providerID, longReserved) + + handle, err := s.openProviderTunnel(selected.entry, tunnelReq, req, queueReason, true) + if err != nil { + s.queue.releaseRun(runID, "send-error") + return nil, err + } + return handle, nil +} + +func (s *Service) submitProviderTunnelDirect(req SubmitProviderTunnelRequest) (ProviderTunnelResult, error) { + entry, err := s.ResolveNode(req.NodeRef) + if err != nil { + return nil, err + } + tunnelReq, _, err := buildProviderTunnelRequest(req, req.Adapter, req.Target) + if err != nil { + return nil, err + } + return s.openProviderTunnel(entry, tunnelReq, req, "dispatched", false) +} + +// openProviderTunnel subscribes the request-bound frame channel, sends the +// tunnel request to the node, and wraps the raw channel so the reserved +// provider-pool slot is released exactly once on END/ERROR/close. +func (s *Service) openProviderTunnel(entry *edgenode.NodeEntry, tunnelReq *iop.ProviderTunnelRequest, req SubmitProviderTunnelRequest, queueReason string, tracked bool) (*ProviderTunnelHandle, error) { + raw, unsubscribe := s.tunnels.subscribe(tunnelReq.GetTunnelId(), tunnelFrameBuffer) + if err := entry.Client.Send(tunnelReq); err != nil { + unsubscribe() + return nil, err + } + + runID := tunnelReq.GetRunId() + var releaseOnce sync.Once + release := func(reason string) { + releaseOnce.Do(func() { + if tracked && s.queue != nil { + s.queue.releaseRun(runID, reason) + } + }) + } + + out := make(chan *iop.ProviderTunnelFrame, tunnelFrameBuffer) + done := make(chan struct{}) + go func() { + defer close(out) + for { + select { + case frame, ok := <-raw: + if !ok { + release("tunnel-stream-closed") + return + } + select { + case out <- frame: + case <-done: + release("tunnel-closed") + return + } + if isTerminalProviderTunnelFrame(frame) { + release("tunnel-" + frame.GetKind().String()) + return + } + case <-done: + release("tunnel-closed") + return + } + } + }() + + return &ProviderTunnelHandle{ + RunDispatch: RunDispatch{ + RunID: runID, + NodeID: entry.NodeID, + NodeLabel: nodeLabel(entry), + ModelGroupKey: req.ModelGroupKey, + Adapter: tunnelReq.GetAdapter(), + Target: tunnelReq.GetTarget(), + SessionID: tunnelReq.GetSessionId(), + TimeoutSec: int(tunnelReq.GetTimeoutSec()), + EstimatedInputTokens: req.EstimatedInputTokens, + ContextClass: req.ContextClass, + QueueReason: queueReason, + }, + TunnelID: tunnelReq.GetTunnelId(), + frames: out, + close: func() { + close(done) + unsubscribe() + release("tunnel-closed") + }, + }, nil +} + +func isTerminalProviderTunnelFrame(f *iop.ProviderTunnelFrame) bool { + switch f.GetKind() { + case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, + iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR: + return true + default: + return false + } +} + +func buildProviderTunnelRequest(req SubmitProviderTunnelRequest, adapter, target string) (*iop.ProviderTunnelRequest, string, error) { + runID := req.RunID + if runID == "" { + runID = NewRunID() + } + body := req.Body + if req.BuildBody != nil { + b, err := req.BuildBody(target) + if err != nil { + return nil, "", err + } + body = b + } + headers := make(map[string]string, len(req.Headers)) + for k, v := range req.Headers { + headers[k] = v + } + metadata := make(map[string]string, len(req.Metadata)) + for k, v := range req.Metadata { + metadata[k] = v + } + return &iop.ProviderTunnelRequest{ + RunId: runID, + TunnelId: runID + "-tunnel", + Adapter: adapter, + Target: target, + Method: req.Method, + Path: req.Path, + Headers: headers, + Body: body, + Stream: req.Stream, + TimeoutSec: int32(normalizeTimeoutSec(req.TimeoutSec)), + Metadata: metadata, + SessionId: NormalizeSessionID(req.SessionID), + }, runID, nil +} + type CancelRunRequest struct { NodeRef string RunID string diff --git a/apps/edge/internal/service/run_dispatch_internal_test.go b/apps/edge/internal/service/run_dispatch_internal_test.go index d915f06..af338c0 100644 --- a/apps/edge/internal/service/run_dispatch_internal_test.go +++ b/apps/edge/internal/service/run_dispatch_internal_test.go @@ -1,6 +1,21 @@ package service -import "testing" +import ( + "context" + "net" + "strings" + "sync" + "testing" + "time" + + toki "git.toki-labs.com/toki/proto-socket/go" + "google.golang.org/protobuf/proto" + + edgeevents "iop/apps/edge/internal/events" + edgenode "iop/apps/edge/internal/node" + "iop/packages/go/config" + iop "iop/proto/gen/iop" +) func TestRunHandleCloseIsIdempotent(t *testing.T) { calls := 0 @@ -17,3 +32,292 @@ func TestRunHandleCloseIsIdempotent(t *testing.T) { t.Fatalf("close called %d times, want 1", calls) } } + +func waitForCondition(t *testing.T, cond func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal(msg) +} + +func inflightRunCount(q *modelQueueManager) int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.inflightByRun) +} + +// TestRouteProviderTunnelFrameDoesNotPublishToBus verifies that tunnel frames +// reach only the request-bound tunnel subscriber and never the run event bus +// fanout (negative test for the passthrough/no-sideband boundary). +func TestRouteProviderTunnelFrameDoesNotPublishToBus(t *testing.T) { + bus := edgeevents.NewBus() + svc := New(edgenode.NewRegistry(), bus) + + runCh, unsubRun := bus.SubscribeAllRuns(8) + defer unsubRun() + + tunnelCh, unsubTunnel := svc.tunnels.subscribe("tunnel-neg-1", 8) + defer unsubTunnel() + + frame := &iop.ProviderTunnelFrame{ + RunId: "run-neg-1", + TunnelId: "tunnel-neg-1", + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, + Body: []byte("data: raw provider bytes\n\n"), + } + svc.RouteProviderTunnelFrame(frame) + + select { + case got := <-tunnelCh: + if string(got.GetBody()) != string(frame.GetBody()) { + t.Fatalf("tunnel frame body mismatch: %q", got.GetBody()) + } + case <-time.After(time.Second): + t.Fatal("tunnel subscriber did not receive the frame") + } + + select { + case e := <-runCh: + t.Fatalf("tunnel frame leaked into run event bus: %+v", e) + case <-time.After(100 * time.Millisecond): + } +} + +// TestRouteProviderTunnelFrameWithoutSubscriberIsDropped verifies routing a +// frame with no subscriber neither blocks nor panics. +func TestRouteProviderTunnelFrameWithoutSubscriberIsDropped(t *testing.T) { + svc := New(edgenode.NewRegistry(), nil) + svc.RouteProviderTunnelFrame(&iop.ProviderTunnelFrame{TunnelId: "no-sub"}) +} + +type providerTunnelTestEnv struct { + svc *Service + nodeClient *toki.TcpClient + capturedMu sync.Mutex + captured *iop.ProviderTunnelRequest +} + +func (e *providerTunnelTestEnv) capturedRequest() *iop.ProviderTunnelRequest { + e.capturedMu.Lock() + defer e.capturedMu.Unlock() + return e.captured +} + +// newProviderTunnelTestEnv wires a provider-pool service to a fake node over +// net.Pipe and captures the ProviderTunnelRequest the node receives. +func newProviderTunnelTestEnv(t *testing.T) *providerTunnelTestEnv { + t.Helper() + edgeConn, nodeConn := net.Pipe() + t.Cleanup(func() { + _ = edgeConn.Close() + _ = nodeConn.Close() + }) + + parserMap := toki.ParserMap{ + toki.TypeNameOf(&iop.ProviderTunnelRequest{}): func(b []byte) (proto.Message, error) { + m := &iop.ProviderTunnelRequest{} + return m, proto.Unmarshal(b, m) + }, + } + edgeClient := toki.NewTcpClient(edgeConn, 0, 0, parserMap) + nodeClient := toki.NewTcpClient(nodeConn, 0, 0, parserMap) + + env := &providerTunnelTestEnv{nodeClient: nodeClient} + toki.AddListenerTyped[*iop.ProviderTunnelRequest](&nodeClient.Communicator, func(req *iop.ProviderTunnelRequest) { + env.capturedMu.Lock() + env.captured = req + env.capturedMu.Unlock() + }) + + store := edgenode.NewNodeStore() + store.Add(&edgenode.NodeRecord{ + ID: "node-pool", + Runtime: config.RuntimeConf{Concurrency: 4}, + Adapters: config.AdaptersConf{ + VllmInstances: []config.VllmInstanceConf{ + {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, + }, + }, + Providers: []config.NodeProviderConf{ + { + ID: "prov-vllm-01", + Adapter: "vllm-gpu", + Models: []string{"served-qwen"}, + Health: "available", + Capacity: 1, + }, + }, + }) + + reg := edgenode.NewRegistry() + reg.Register(&edgenode.NodeEntry{ + NodeID: "node-pool", + LifecycleState: edgenode.LifecycleConnected, + Client: edgeClient, + }) + + svc := New(reg, edgeevents.NewBus()) + svc.SetNodeStore(store) + svc.SetModelCatalog([]config.ModelCatalogEntry{ + {ID: "qwen3.6:35b", Providers: map[string]string{"prov-vllm-01": "served-qwen"}}, + }) + env.svc = svc + return env +} + +// TestSubmitProviderTunnelProviderPoolSendsRequestAndReleasesSlotOnEnd covers +// the provider-pool tunnel path end to end: admission rewrites adapter/target, +// BuildBody receives the served target, ordered frames reach the handle, and +// the admission slot is released on the END frame. +func TestSubmitProviderTunnelProviderPoolSendsRequestAndReleasesSlotOnEnd(t *testing.T) { + env := newProviderTunnelTestEnv(t) + svc := env.svc + + var builtTarget string + handle, err := svc.SubmitProviderTunnel(context.Background(), SubmitProviderTunnelRequest{ + RunID: "run-tunnel-001", + ModelGroupKey: "qwen3.6:35b", + Method: "POST", + Path: "/v1/chat/completions", + BuildBody: func(target string) ([]byte, error) { + builtTarget = target + return []byte(`{"model":"` + target + `","stream":true}`), nil + }, + Stream: true, + ProviderPool: true, + }) + if err != nil { + t.Fatalf("SubmitProviderTunnel: %v", err) + } + defer handle.Close() + + if builtTarget != "served-qwen" { + t.Errorf("BuildBody target: got %q, want served-qwen", builtTarget) + } + dispatch := handle.Dispatch() + if dispatch.Adapter != "vllm-gpu" || dispatch.Target != "served-qwen" { + t.Errorf("dispatch adapter/target: got %q/%q", dispatch.Adapter, dispatch.Target) + } + + waitForCondition(t, func() bool { return env.capturedRequest() != nil }, + "fake node did not receive ProviderTunnelRequest") + captured := env.capturedRequest() + if captured.GetAdapter() != "vllm-gpu" || captured.GetTarget() != "served-qwen" { + t.Errorf("wire adapter/target: got %q/%q", captured.GetAdapter(), captured.GetTarget()) + } + if !strings.Contains(string(captured.GetBody()), `"model":"served-qwen"`) { + t.Errorf("wire body missing served model: %s", captured.GetBody()) + } + if inflightRunCount(svc.queue) != 1 { + t.Fatalf("expected 1 inflight run after dispatch, got %d", inflightRunCount(svc.queue)) + } + + tunnelID := captured.GetTunnelId() + frames := []*iop.ProviderTunnelFrame{ + {RunId: captured.GetRunId(), TunnelId: tunnelID, Sequence: 0, Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START, StatusCode: 200}, + {RunId: captured.GetRunId(), TunnelId: tunnelID, Sequence: 1, Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, Body: []byte("data: chunk-1\n\n")}, + {RunId: captured.GetRunId(), TunnelId: tunnelID, Sequence: 2, Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END, End: true}, + } + for _, f := range frames { + svc.RouteProviderTunnelFrame(f) + } + + stream := handle.Stream() + for i, want := range frames { + select { + case got := <-stream.Frames: + if got.GetSequence() != want.GetSequence() || got.GetKind() != want.GetKind() { + t.Fatalf("frame %d: got seq=%d kind=%v", i, got.GetSequence(), got.GetKind()) + } + case <-time.After(time.Second): + t.Fatalf("frame %d not delivered", i) + } + } + select { + case _, open := <-stream.Frames: + if open { + t.Fatal("expected frame channel to close after END") + } + case <-time.After(time.Second): + t.Fatal("frame channel not closed after END") + } + + waitForCondition(t, func() bool { return inflightRunCount(svc.queue) == 0 }, + "admission slot not released after END frame") +} + +// TestSubmitProviderTunnelCloseReleasesSlot verifies the cancel path: closing +// the handle before a terminal frame releases the provider-pool slot. +func TestSubmitProviderTunnelCloseReleasesSlot(t *testing.T) { + env := newProviderTunnelTestEnv(t) + svc := env.svc + + handle, err := svc.SubmitProviderTunnel(context.Background(), SubmitProviderTunnelRequest{ + RunID: "run-tunnel-cancel", + ModelGroupKey: "qwen3.6:35b", + Method: "POST", + Path: "/v1/chat/completions", + Body: []byte(`{"model":"qwen3.6:35b"}`), + ProviderPool: true, + }) + if err != nil { + t.Fatalf("SubmitProviderTunnel: %v", err) + } + if inflightRunCount(svc.queue) != 1 { + t.Fatalf("expected 1 inflight run after dispatch, got %d", inflightRunCount(svc.queue)) + } + + handle.Close() + + waitForCondition(t, func() bool { return inflightRunCount(svc.queue) == 0 }, + "admission slot not released after Close") +} + +// TestSubmitProviderTunnelErrorFrameReleasesSlot verifies the ERROR frame is +// terminal and releases the provider-pool slot like END. +func TestSubmitProviderTunnelErrorFrameReleasesSlot(t *testing.T) { + env := newProviderTunnelTestEnv(t) + svc := env.svc + + handle, err := svc.SubmitProviderTunnel(context.Background(), SubmitProviderTunnelRequest{ + RunID: "run-tunnel-error", + ModelGroupKey: "qwen3.6:35b", + Method: "POST", + Path: "/v1/chat/completions", + Body: []byte(`{"model":"qwen3.6:35b"}`), + ProviderPool: true, + }) + if err != nil { + t.Fatalf("SubmitProviderTunnel: %v", err) + } + defer handle.Close() + + waitForCondition(t, func() bool { return env.capturedRequest() != nil }, + "fake node did not receive ProviderTunnelRequest") + captured := env.capturedRequest() + + svc.RouteProviderTunnelFrame(&iop.ProviderTunnelFrame{ + RunId: captured.GetRunId(), + TunnelId: captured.GetTunnelId(), + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR, + Error: "provider unavailable", + }) + + stream := handle.Stream() + select { + case got := <-stream.Frames: + if got.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR { + t.Fatalf("expected ERROR frame, got %v", got.GetKind()) + } + case <-time.After(time.Second): + t.Fatal("ERROR frame not delivered") + } + + waitForCondition(t, func() bool { return inflightRunCount(svc.queue) == 0 }, + "admission slot not released after ERROR frame") +} diff --git a/apps/edge/internal/service/service.go b/apps/edge/internal/service/service.go index 6304627..016231d 100644 --- a/apps/edge/internal/service/service.go +++ b/apps/edge/internal/service/service.go @@ -26,10 +26,11 @@ type Service struct { nodeStore *edgenode.NodeStore queue *modelQueueManager modelCatalog []config.ModelCatalogEntry + tunnels *providerTunnelRouter } func New(registry *edgenode.Registry, events *edgeevents.Bus) *Service { - s := &Service{registry: registry, events: events} + s := &Service{registry: registry, events: events, tunnels: newProviderTunnelRouter()} if events != nil { q := newModelQueueManager(nil) q.startEventWatcher(events) diff --git a/apps/edge/internal/transport/server.go b/apps/edge/internal/transport/server.go index f11eb78..9a465f0 100644 --- a/apps/edge/internal/transport/server.go +++ b/apps/edge/internal/transport/server.go @@ -67,6 +67,7 @@ type Server struct { handlerMu sync.RWMutex onRunEvent func(*iop.RunEvent) onNodeEvent func(*iop.EdgeNodeEvent) + onTunnelFrame func(*iop.ProviderTunnelFrame) stopping atomic.Bool HeartbeatInterval int HeartbeatWait int @@ -135,6 +136,22 @@ func (s *Server) SetNodeEventHandler(handler func(*iop.EdgeNodeEvent)) { s.handlerMu.Unlock() } +// SetTunnelFrameHandler registers the request-bound ProviderTunnelFrame +// handler. Tunnel frames carry raw provider passthrough bytes and are routed +// to a per-request channel by the handler; they must never be published to +// the run event bus. +func (s *Server) SetTunnelFrameHandler(handler func(*iop.ProviderTunnelFrame)) { + s.handlerMu.Lock() + s.onTunnelFrame = handler + s.handlerMu.Unlock() +} + +func (s *Server) HasTunnelFrameHandler() bool { + s.handlerMu.RLock() + defer s.handlerMu.RUnlock() + return s.onTunnelFrame != nil +} + func (s *Server) HasRunEventHandler() bool { s.handlerMu.RLock() defer s.handlerMu.RUnlock() @@ -164,6 +181,20 @@ func (s *Server) onNodeConnected(client *toki.TcpClient) { } }) + toki.AddListenerTyped[*iop.ProviderTunnelFrame](&client.Communicator, func(f *iop.ProviderTunnelFrame) { + s.handlerMu.RLock() + handler := s.onTunnelFrame + s.handlerMu.RUnlock() + if handler == nil { + s.logger.Warn("provider tunnel frame dropped: no tunnel handler", + zap.String("run_id", f.GetRunId()), + zap.String("tunnel_id", f.GetTunnelId()), + ) + return + } + handler(f) + }) + toki.AddRequestListenerTyped[*iop.RegisterRequest, *iop.RegisterResponse]( &client.Communicator, func(req *iop.RegisterRequest) (*iop.RegisterResponse, error) { diff --git a/apps/edge/internal/transport/server_test.go b/apps/edge/internal/transport/server_test.go index 8defac4..abe5fdb 100644 --- a/apps/edge/internal/transport/server_test.go +++ b/apps/edge/internal/transport/server_test.go @@ -2,7 +2,10 @@ package transport import ( "context" + "net" + "sync" "testing" + "time" toki "git.toki-labs.com/toki/proto-socket/go" "go.uber.org/zap" @@ -194,6 +197,81 @@ func TestEdgeParserMap_ProviderTunnelFrameIsSeparateFromRunEvent(t *testing.T) { } } +// TestServerRoutesTunnelFramesToTunnelHandlerNotRunHandler verifies that a +// ProviderTunnelFrame received from a node connection reaches only the tunnel +// frame handler: the run event handler (events.Bus publisher in production) +// must never observe tunnel frames, and run events must not reach the tunnel +// handler. +func TestServerRoutesTunnelFramesToTunnelHandlerNotRunHandler(t *testing.T) { + edgeConn, nodeConn := net.Pipe() + defer edgeConn.Close() + defer nodeConn.Close() + + edgeClient := toki.NewTcpClient(edgeConn, 0, 0, edgeParserMap()) + nodeClient := toki.NewTcpClient(nodeConn, 0, 0, toki.ParserMap{}) + + s := &Server{ + registry: edgenode.NewRegistry(), + logger: zap.NewNop(), + } + + var mu sync.Mutex + var tunnelFrames []*iop.ProviderTunnelFrame + var runEvents []*iop.RunEvent + s.SetTunnelFrameHandler(func(f *iop.ProviderTunnelFrame) { + mu.Lock() + tunnelFrames = append(tunnelFrames, f) + mu.Unlock() + }) + s.SetRunEventHandler(func(e *iop.RunEvent) { + mu.Lock() + runEvents = append(runEvents, e) + mu.Unlock() + }) + + s.onNodeConnected(edgeClient) + + rawBody := []byte("data: {\"choices\":[{\"delta\":{\"content\":\"raw\"}}]}\n\n") + if err := nodeClient.Send(&iop.ProviderTunnelFrame{ + RunId: "run-1", + TunnelId: "tunnel-1", + Sequence: 1, + Kind: iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY, + Body: rawBody, + }); err != nil { + t.Fatalf("send tunnel frame: %v", err) + } + if err := nodeClient.Send(&iop.RunEvent{RunId: "run-1", Type: "delta", Delta: "normalized"}); err != nil { + t.Fatalf("send run event: %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + mu.Lock() + done := len(tunnelFrames) == 1 && len(runEvents) == 1 + mu.Unlock() + if done { + break + } + time.Sleep(10 * time.Millisecond) + } + + mu.Lock() + defer mu.Unlock() + if len(tunnelFrames) != 1 { + t.Fatalf("tunnel handler frames: got %d want 1", len(tunnelFrames)) + } + if string(tunnelFrames[0].GetBody()) != string(rawBody) { + t.Fatalf("tunnel frame body mismatch: %q", tunnelFrames[0].GetBody()) + } + if len(runEvents) != 1 { + t.Fatalf("run handler events: got %d want 1", len(runEvents)) + } + if runEvents[0].GetDelta() != "normalized" { + t.Fatalf("run handler must only see run events, got %+v", runEvents[0]) + } +} + func TestServerEnrichesRunEventNodeAlias(t *testing.T) { registry := edgenode.NewRegistry() registry.Register(&edgenode.NodeEntry{NodeID: "node-1", Alias: "alias-1"})