From bc966f8cb364bea801151e0432c995308e251d13 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 8 Jul 2026 10:21:10 +0900 Subject: [PATCH] sync: routing policy model orchestration updates and node implementation --- ...patible-raw-tunnel-sideband-passthrough.md | 24 ++++--- .../SDD.md | 0 .../PHASE.md | 4 +- apps/node/internal/node/node.go | 17 ++++- apps/node/internal/node/node_test.go | 70 +++++++++++++++++++ 5 files changed, 99 insertions(+), 16 deletions(-) rename agent-roadmap/{ => archive}/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md (86%) rename agent-roadmap/{ => archive}/sdd/routing-policy-model-orchestration/openai-compatible-raw-tunnel-sideband-passthrough/SDD.md (100%) diff --git a/agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md b/agent-roadmap/archive/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md similarity index 86% rename from agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md rename to agent-roadmap/archive/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md index 0727f0d..0a5b267 100644 --- a/agent-roadmap/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md +++ b/agent-roadmap/archive/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md @@ -13,7 +13,7 @@ OpenAI-compatible 추론 엔진(vLLM, SGLang, Lemonade, OpenAI-compatible API pr ## 상태 -[진행중] +[완료] ## 승격 조건 @@ -68,22 +68,24 @@ OpenAI-compatible provider의 원본 응답을 정규화/재조립하지 않고 raw bytes는 전송 source of truth로 쓰되, 운영 관측과 저장은 사람이 읽을 수 있는 조립 결과와 구조화된 sideband를 기준으로 한다. -- [ ] [assembled-log] 실행 로그에는 raw chunk payload 대신 assembled `content`, `reasoning`, `tool_calls`, sideband summary, transformed output을 남긴다. 검증: log fixture가 readable assembled string과 구조화 관측값을 포함하고 raw body chunk를 기본 저장하지 않는다. -- [ ] [backpressure-cancel] tunnel path는 기존 socket 위에서 ordered, lossless, backpressure-aware stream을 유지하고 cancel/control frame을 지연시키지 않는다. 검증: slow client, cancel, provider error test에서 chunk drop/reorder 없이 upstream request가 중단된다. +- [x] [assembled-log] 실행 로그에는 raw chunk payload 대신 assembled `content`, `reasoning`, `tool_calls`, sideband summary, transformed output을 남긴다. 검증: log fixture가 readable assembled string과 구조화 관측값을 포함하고 raw body chunk를 기본 저장하지 않는다. +- [x] [backpressure-cancel] tunnel path는 기존 socket 위에서 ordered, lossless, backpressure-aware stream을 유지하고 cancel/control frame을 지연시키지 않는다. 검증: slow client, cancel, provider error test에서 chunk drop/reorder 없이 upstream request가 중단된다. - [x] [contract-docs] agent-contract, agent-spec, config/docs 예시에 raw tunnel, response mode, sideband/transformed semantics, legacy normalized path migration을 반영한다. 검증: contract/spec 링크와 코드 test fixture가 같은 mode 이름과 default를 사용한다. ## 완료 리뷰 -- 상태: 없음 -- 요청일: 없음 -- 완료 근거: 기능 Task가 아직 충족되지 않았다. +- 상태: 통과 +- 요청일: 2026-07-08 +- 완료 근거: 기능 Task 9개와 SDD S01-S11 evidence를 같은 milestone task group의 archive `complete.log` 5개, 코드 감사, 최종 검증으로 확인했다. - 검토 항목: - - [ ] 구현 결과가 pure `passthrough` 기본값과 provider-original byte stream을 보장한다. - - [ ] sideband/transformed mode가 opt-in으로만 동작한다. - - [ ] raw tunnel이 기존 Edge-Node socket 위에서 lossless/backpressure-aware하게 동작한다. - - [ ] archive 이동을 승인했다. + - [x] 구현 결과가 pure `passthrough` 기본값과 provider-original byte stream을 보장한다. + - [x] sideband/transformed mode가 opt-in으로만 동작한다. + - [x] raw tunnel이 기존 Edge-Node socket 위에서 lossless/backpressure-aware하게 동작한다. + - [x] archive 이동을 승인했다. - agent-ui 상태 반영: 해당 없음 -- 리뷰 코멘트: 없음 +- Spec sync: not-needed - [openai-compatible-surface.md](../../../../../agent-spec/input/openai-compatible-surface.md)와 [edge-node-execution.md](../../../../../agent-spec/runtime/edge-node-execution.md)는 raw tunnel, response mode, cancel 경계를 이미 반영하며 이번 보완은 spec/contract와 구현을 맞춘 코드 수정이다. +- Workspace 잠금: 관련 lock 없음 +- 리뷰 코멘트: 코드 감사 중 Node provider tunnel이 `CancelRequest(CANCEL_RUN)`으로 취소되지 않는 S09 회귀를 발견해 `apps/node/internal/node/node.go`와 `apps/node/internal/node/node_test.go`에서 즉시 보완했다. `go test ./...`, targeted Node/Edge/race 테스트, `make test-e2e`, `git diff --check`로 종료 가능 상태를 확인했다. ## 범위 제외 diff --git a/agent-roadmap/sdd/routing-policy-model-orchestration/openai-compatible-raw-tunnel-sideband-passthrough/SDD.md b/agent-roadmap/archive/sdd/routing-policy-model-orchestration/openai-compatible-raw-tunnel-sideband-passthrough/SDD.md similarity index 100% rename from agent-roadmap/sdd/routing-policy-model-orchestration/openai-compatible-raw-tunnel-sideband-passthrough/SDD.md rename to agent-roadmap/archive/sdd/routing-policy-model-orchestration/openai-compatible-raw-tunnel-sideband-passthrough/SDD.md diff --git a/agent-roadmap/phase/routing-policy-model-orchestration/PHASE.md b/agent-roadmap/phase/routing-policy-model-orchestration/PHASE.md index 02fc68f..77300b1 100644 --- a/agent-roadmap/phase/routing-policy-model-orchestration/PHASE.md +++ b/agent-roadmap/phase/routing-policy-model-orchestration/PHASE.md @@ -16,8 +16,8 @@ IOP의 OpenAI-compatible, A2A, IOP native 입력 표면에서 들어온 요청 완료, 검토중, 진행중, 계획, 스케치 순서로 두어 아래로 갈수록 미래 작업에 가까워지게 정렬한다. 스케치 Milestone은 아직 구현 가능한 계획이 아니므로 계획 Milestone보다 아래에 둔다. -- [진행중] OpenAI-compatible Raw Tunnel과 Sideband Passthrough - - 경로: [openai-compatible-raw-tunnel-sideband-passthrough](milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) +- [완료] OpenAI-compatible Raw Tunnel과 Sideband Passthrough + - 경로: [openai-compatible-raw-tunnel-sideband-passthrough](../../archive/phase/routing-policy-model-orchestration/milestones/openai-compatible-raw-tunnel-sideband-passthrough.md) - 요약: OpenAI-compatible provider 응답을 기존 Edge-Node proto-socket 위 lossless raw tunnel로 전달하고, 기본값은 provider-original `passthrough`로 두며, 요청이 명시한 경우에만 `passthrough+sideband` 또는 `transformed`를 사용한다. - [스케치] OpenAI-compatible 하이브리드 라우팅과 컨텍스트 최적화 diff --git a/apps/node/internal/node/node.go b/apps/node/internal/node/node.go index 8e17202..e183dad 100644 --- a/apps/node/internal/node/node.go +++ b/apps/node/internal/node/node.go @@ -269,12 +269,23 @@ func (n *Node) OnProviderTunnelRequest(ctx context.Context, sess *transport.Sess nodeAlias: nodeAlias, } - execCtx := ctx + execCtx, cancel := context.WithCancel(ctx) if tr.TimeoutSec > 0 { - var cancel context.CancelFunc execCtx, cancel = context.WithTimeout(ctx, time.Duration(tr.TimeoutSec)*time.Second) - defer cancel() } + defer cancel() + + h := &runHandle{ + runID: tr.RunID, + adapter: tr.Adapter, + target: tr.Target, + sessionID: normalizeSessionID(tr.SessionID), + cancel: cancel, + done: make(chan struct{}), + } + n.runs.register(h) + defer n.runs.deregister(tr.RunID) + defer close(h.done) if err := tunnelAdapter.TunnelProvider(execCtx, tr, sink); err != nil { n.logger.Warn("provider tunnel error", diff --git a/apps/node/internal/node/node_test.go b/apps/node/internal/node/node_test.go index 2f3bc85..6c30cbd 100644 --- a/apps/node/internal/node/node_test.go +++ b/apps/node/internal/node/node_test.go @@ -3077,6 +3077,23 @@ func (a *mockTunnelAdapter) TunnelProvider(ctx context.Context, req runtime.Prov return err } +type cancelAwareTunnelAdapter struct { + countingAdapter + started chan struct{} + observedCancel chan struct{} +} + +func (a *cancelAwareTunnelAdapter) Name() string { return "openai_compat" } +func (a *cancelAwareTunnelAdapter) Capabilities(_ context.Context) (runtime.Capabilities, error) { + return runtime.Capabilities{AdapterName: "openai_compat", Targets: []string{"qwen"}}, nil +} +func (a *cancelAwareTunnelAdapter) TunnelProvider(ctx context.Context, _ runtime.ProviderTunnelRequest, _ runtime.ProviderTunnelSink) error { + close(a.started) + <-ctx.Done() + close(a.observedCancel) + return ctx.Err() +} + func TestNodeOnProviderTunnelRequest_Success(t *testing.T) { mta := &mockTunnelAdapter{ t: t, @@ -3104,6 +3121,59 @@ func TestNodeOnProviderTunnelRequest_Success(t *testing.T) { } } +func TestNodeOnProviderTunnelRequest_CancelRequestCancelsProviderContext(t *testing.T) { + adapter := &cancelAwareTunnelAdapter{ + started: make(chan struct{}), + observedCancel: make(chan struct{}), + } + router := &fixedRouter{adapterName: "openai_compat", adapters: map[string]runtime.Adapter{"openai_compat": adapter}} + n, _ := makeNode(t, router) + + req := &iop.ProviderTunnelRequest{ + RunId: "run-tunnel-cancel", + TunnelId: "tunnel-cancel", + Adapter: "openai_compat", + Target: "qwen", + Method: "POST", + Path: "/v1/chat/completions", + TimeoutSec: 30, + } + + errCh := make(chan error, 1) + go func() { + errCh <- n.OnProviderTunnelRequest(context.Background(), nil, req) + }() + + select { + case <-adapter.started: + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for tunnel adapter to start") + } + + err := n.OnCancel(context.Background(), nil, &iop.CancelRequest{ + RunId: "run-tunnel-cancel", + Action: iop.CancelAction_CANCEL_ACTION_CANCEL_RUN, + }) + if err != nil { + t.Fatalf("OnCancel failed: %v", err) + } + + select { + case <-adapter.observedCancel: + case <-time.After(2 * time.Second): + t.Fatal("tunnel adapter did not observe cancel request") + } + + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("OnProviderTunnelRequest error = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for tunnel request to finish after cancel") + } +} + func buildSessionTestPipeForNode(t *testing.T) (edgeSide *toki.TcpClient, sess *transport.Session) { t.Helper() edgeConn, nodeConn := net.Pipe()