From 909f916df8ba90ddcff38150da2d61e3f7b8f2e5 Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 8 Jul 2026 16:14:14 +0900 Subject: [PATCH] =?UTF-8?q?fix(openai):=20passthrough=20=EC=9D=91=EB=8B=B5?= =?UTF-8?q?=20=EB=AA=A8=EB=8D=B8=20alias=EB=A5=BC=20=EB=B3=B4=EC=A1=B4?= =?UTF-8?q?=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent-contract/outer/openai-compatible-api.md | 2 +- apps/edge/internal/openai/server_test.go | 103 ++++++++++- apps/edge/internal/openai/stream.go | 162 +++++++++++++++++- 3 files changed, 253 insertions(+), 14 deletions(-) diff --git a/agent-contract/outer/openai-compatible-api.md b/agent-contract/outer/openai-compatible-api.md index bad71f1..96e7d48 100644 --- a/agent-contract/outer/openai-compatible-api.md +++ b/agent-contract/outer/openai-compatible-api.md @@ -174,7 +174,7 @@ Workspace-bound route는 workspace가 없거나 상대 경로이면 OpenAI-compa OpenAI-compatible inference provider route는 요청 metadata의 `iop_response_mode`로 응답 경로를 고른다. -- `metadata.iop_response_mode` 생략 또는 `passthrough`: provider HTTP status/header/body bytes를 Node가 열어 기존 Edge-Node tunnel로 relay하고, Edge가 caller에게 그대로 쓴다. pure `passthrough` 응답 body에는 IOP sideband field/event를 섞지 않고, `X-IOP-Response-Mode` header도 붙이지 않는다. +- `metadata.iop_response_mode` 생략 또는 `passthrough`: provider HTTP status/header/body를 Node가 열어 기존 Edge-Node tunnel로 relay하고, Edge가 caller에게 쓴다. Chat Completions 성공 응답의 top-level `model` echo가 provider-served model이면 caller가 요청한 IOP model alias로 정규화한다. reasoning/content/tool_calls 같은 provider payload field는 보존한다. pure `passthrough` 응답 body에는 IOP sideband field/event를 섞지 않고, `X-IOP-Response-Mode` header도 붙이지 않는다. - `metadata.iop_response_mode="passthrough+sideband"`: provider body와 IOP route/usage/assembled observation을 명시적 IOP extension surface로 함께 노출한다. streaming 응답은 provider SSE event 경계 사이에 `event: iop.sideband`를 추가하고, non-streaming 응답은 `iop.chat.passthrough_sideband` envelope로 provider body와 `iop_sideband`를 함께 반환한다. 이 모드는 provider-original byte-identical response로 표시하지 않는다. - `metadata.iop_response_mode="transformed"`: raw tunnel을 쓰지 않고 normalized IOP output path를 사용하며 `X-IOP-Response-Mode: transformed`로 라벨링한다. 이 응답은 provider-original byte identity 검증 대상이 아니다. diff --git a/apps/edge/internal/openai/server_test.go b/apps/edge/internal/openai/server_test.go index b22c76b..bab3f2c 100644 --- a/apps/edge/internal/openai/server_test.go +++ b/apps/edge/internal/openai/server_test.go @@ -4363,8 +4363,8 @@ func TestChatCompletionsPassthroughSidebandNonStreamingWrapsProviderBody(t *test } // TestChatCompletionsPassthroughDoesNotExposeSideband verifies SDD S06: pure -// passthrough (omitted mode) stays byte-identical even when the tunnel emits -// usage frames, and no IOP sideband marker, event, or label reaches the caller. +// passthrough (omitted mode) does not expose usage frames, IOP sideband marker, +// event, or label to the caller. func TestChatCompletionsPassthroughDoesNotExposeSideband(t *testing.T) { providerBody := "data: {\"choices\":[{\"delta\":{\"content\":\"pure\"}}]}\n\ndata: [DONE]\n\n" frames := make(chan *iop.ProviderTunnelFrame, 4) @@ -4449,8 +4449,7 @@ func TestChatCompletionsTransformedModeLabelsIOPOutput(t *testing.T) { } // TestChatCompletionsPassthroughDoesNotLabelTransformed verifies pure -// passthrough responses carry no IOP transformed label and stay byte-identical -// to the provider body. +// passthrough responses carry no IOP transformed label. func TestChatCompletionsPassthroughDoesNotLabelTransformed(t *testing.T) { providerBody := `{"ok":true}` frames := make(chan *iop.ProviderTunnelFrame, 3) @@ -4541,10 +4540,59 @@ func TestChatCompletionsPassthroughNonStreamingByteIdentity(t *testing.T) { } } +func TestChatCompletionsPassthroughNonStreamingRewritesModelEcho(t *testing.T) { + providerBody := `{"id":"cmpl-provider-1","object":"chat.completion","created":123,"model":"served-model","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.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", + "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()) + } + var resp struct { + Model string `json:"model"` + Choices []struct { + Message struct { + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + } `json:"message"` + } `json:"choices"` + ProviderExtra map[string]int `json:"provider_extra"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode response: %v body=%s", err, w.Body.String()) + } + if resp.Model != "pool-model" { + t.Fatalf("response model: got %q, want pool-model; body=%s", resp.Model, w.Body.String()) + } + if resp.Choices[0].Message.Content != "hi" || resp.Choices[0].Message.ReasoningContent != "because" { + t.Fatalf("provider content/reasoning not preserved: %+v", resp.Choices[0].Message) + } + if resp.ProviderExtra["a"] != 1 { + t.Fatalf("provider_extra not preserved: %+v", resp.ProviderExtra) + } + if !strings.Contains(string(gotProviderReq), `"model":"served-model"`) { + t.Fatalf("provider request body missing served model: %s", gotProviderReq) + } +} + // 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. +// before dispatching to the provider. 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 @@ -4616,7 +4664,8 @@ func TestChatCompletionsPassthroughProviderPoolGenerationPolicy(t *testing.T) { // 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. +// with provider fields preserved except the top-level model echo, which is +// normalized back to the caller-facing model alias. func TestChatCompletionsPassthroughStreamingByteIdentity(t *testing.T) { providerBody := "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n" + "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"thinking...\"}}]}\n\n" + @@ -4651,6 +4700,48 @@ func TestChatCompletionsPassthroughStreamingByteIdentity(t *testing.T) { } } +func TestChatCompletionsPassthroughStreamingRewritesModelEcho(t *testing.T) { + providerBody := "data: {\"id\":\"chunk-1\",\"object\":\"chat.completion.chunk\",\"created\":123,\"model\":\"served-model\",\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}\n\n" + + "data: {\"id\":\"chunk-1\",\"object\":\"chat.completion.chunk\",\"created\":123,\"model\":\"served-model\",\"choices\":[{\"delta\":{\"reasoning\":\"thinking...\"}}]}\n\n" + + "data: {\"id\":\"chunk-1\",\"object\":\"chat.completion.chunk\",\"created\":123,\"model\":\"served-model\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\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()) + } + body := w.Body.String() + if strings.Contains(body, `"model":"served-model"`) { + t.Fatalf("response leaked provider-served model: %s", body) + } + if strings.Count(body, `"model":"pool-model"`) != 3 { + t.Fatalf("response model aliases not rewritten in every chunk: %s", body) + } + if !strings.Contains(body, `"reasoning":"thinking..."`) || !strings.Contains(body, `"content":"hi"`) { + t.Fatalf("provider reasoning/content not preserved: %s", body) + } + if !strings.Contains(body, "data: [DONE]") { + t.Fatalf("done marker not preserved: %s", body) + } + 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. diff --git a/apps/edge/internal/openai/stream.go b/apps/edge/internal/openai/stream.go index 38f3416..67fa304 100644 --- a/apps/edge/internal/openai/stream.go +++ b/apps/edge/internal/openai/stream.go @@ -308,16 +308,16 @@ 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. +// provider tunnel (SDD S04/S06): provider status, headers, and content are +// relayed without IOP sideband fields or events. Chat Completions model echoes +// are normalized back to the caller-facing model alias. func (s *Server) tunnelChatCompletionPassthrough(w http.ResponseWriter, r *http.Request, req chatCompletionRequest, dispatch routeDispatch, runMeta map[string]string, rawBody []byte, estimate int, contextClass string) { handle, ok := s.submitChatCompletionTunnel(w, r, req, dispatch, runMeta, rawBody, estimate, contextClass, responseModePassthrough) if !ok { return } defer handle.Close() - s.writeProviderTunnelResponse(w, r, handle, req.Stream) + s.writeProviderTunnelResponse(w, r, handle, req.Stream, req.Model) } // tunnelChatCompletionPassthroughSideband serves a Chat Completions request @@ -398,7 +398,7 @@ func (s *Server) submitChatCompletionTunnel(w http.ResponseWriter, r *http.Reque // 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, reqStream bool) { +func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Request, handle edgeservice.ProviderTunnelResult, reqStream bool, requestModel string) { frames := handle.Stream().Frames if frames == nil { writeError(w, http.StatusBadGateway, "provider_tunnel_error", "tunnel stream unavailable") @@ -409,9 +409,12 @@ func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Requ defer timer.Stop() assembler := &providerChatAssembler{streaming: reqStream} + modelRewriter := newProviderModelRewriter(reqStream, requestModel) wroteHeader := false bodyBytes := 0 + var bufferedBody []byte + rewriteModelEcho := modelRewriter != nil defer func() { obs := assembler.observation() s.logger.Info("openai chat completion passthrough closed", @@ -453,6 +456,14 @@ func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Requ if status == 0 { status = http.StatusOK } + if status >= http.StatusBadRequest { + rewriteModelEcho = false + } + if rewriteModelEcho { + // The response body may change length when provider-served + // model names are rewritten to caller-facing model aliases. + w.Header().Del("Content-Length") + } w.WriteHeader(status) wroteHeader = true if flusher != nil { @@ -469,13 +480,24 @@ func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Requ w.WriteHeader(http.StatusOK) wroteHeader = true } + bodyBytes += len(body) + assembler.Write(body) + if rewriteModelEcho { + if reqStream { + body = modelRewriter.AppendStream(body) + } else { + bufferedBody = append(bufferedBody, body...) + continue + } + } + if len(body) == 0 { + continue + } 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) - assembler.Write(body) if flusher != nil { flusher.Flush() } @@ -494,10 +516,35 @@ func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Requ zap.String("run_id", handle.Dispatch().RunID), zap.String("error", msg), ) + if rewriteModelEcho && !reqStream && len(bufferedBody) > 0 { + body := modelRewriter.RewriteComplete(bufferedBody) + _, _ = w.Write(body) + if flusher != nil { + flusher.Flush() + } + } return case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END: if !wroteHeader { writeError(w, http.StatusBadGateway, "provider_tunnel_error", "tunnel ended before provider response") + return + } + if rewriteModelEcho { + var body []byte + if reqStream { + body = modelRewriter.FlushStream() + } else if len(bufferedBody) > 0 { + body = modelRewriter.RewriteComplete(bufferedBody) + } + if len(body) > 0 { + if _, err := w.Write(body); err != nil { + s.sendCancelRun(handle.Dispatch()) + return + } + if flusher != nil { + flusher.Flush() + } + } } return case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_USAGE: @@ -509,6 +556,105 @@ func (s *Server) writeProviderTunnelResponse(w http.ResponseWriter, r *http.Requ } } +type providerModelRewriter struct { + streaming bool + model string + pending []byte +} + +func newProviderModelRewriter(streaming bool, model string) *providerModelRewriter { + model = strings.TrimSpace(model) + if model == "" { + return nil + } + return &providerModelRewriter{streaming: streaming, model: model} +} + +func (r *providerModelRewriter) AppendStream(chunk []byte) []byte { + if r == nil || !r.streaming || len(chunk) == 0 { + return chunk + } + r.pending = append(r.pending, chunk...) + var out bytes.Buffer + for { + idx := bytes.IndexByte(r.pending, '\n') + if idx < 0 { + break + } + line := r.pending[:idx+1] + out.Write(rewriteProviderSSEModelLine(line, r.model)) + r.pending = r.pending[idx+1:] + } + return out.Bytes() +} + +func (r *providerModelRewriter) FlushStream() []byte { + if r == nil || len(r.pending) == 0 { + return nil + } + pending := r.pending + r.pending = nil + return rewriteProviderSSEModelLine(pending, r.model) +} + +func (r *providerModelRewriter) RewriteComplete(body []byte) []byte { + if r == nil || len(body) == 0 { + return body + } + return rewriteProviderJSONModel(body, r.model) +} + +func rewriteProviderSSEModelLine(line []byte, model string) []byte { + body, ending := splitLineEnding(line) + prefix, payload, ok := bytes.Cut(body, []byte(":")) + if !ok || strings.TrimSpace(string(prefix)) != "data" { + return line + } + payload = bytes.TrimSpace(payload) + if len(payload) == 0 || bytes.Equal(payload, []byte("[DONE]")) { + return line + } + rewritten := rewriteProviderJSONModel(payload, model) + if bytes.Equal(rewritten, payload) { + return line + } + out := make([]byte, 0, len("data: ")+len(rewritten)+len(ending)) + out = append(out, "data: "...) + out = append(out, rewritten...) + out = append(out, ending...) + return out +} + +func splitLineEnding(line []byte) ([]byte, []byte) { + if len(line) == 0 || line[len(line)-1] != '\n' { + return line, nil + } + if len(line) >= 2 && line[len(line)-2] == '\r' { + return line[:len(line)-2], []byte("\r\n") + } + return line[:len(line)-1], []byte("\n") +} + +func rewriteProviderJSONModel(body []byte, model string) []byte { + var raw map[string]json.RawMessage + if err := json.Unmarshal(body, &raw); err != nil { + return body + } + if _, ok := raw["model"]; !ok { + return body + } + modelJSON, err := json.Marshal(model) + if err != nil { + return body + } + raw["model"] = modelJSON + rewritten, err := json.Marshal(raw) + if err != nil { + return body + } + return rewritten +} + // Sideband extension surface (SDD S05): explicit passthrough+sideband // responses expose IOP observations alongside provider-original content. // Streaming responses interleave `event: iop.sideband` SSE events at provider @@ -610,6 +756,7 @@ type providerChatAssembler struct { type providerChatDeltaEnvelope struct { Content string `json:"content"` ReasoningContent string `json:"reasoning_content"` + Reasoning string `json:"reasoning"` ToolCalls []struct { Function struct { Name string `json:"name"` @@ -659,6 +806,7 @@ func (a *providerChatAssembler) consumeSSELine(line string) { func (a *providerChatAssembler) consumeDelta(delta providerChatDeltaEnvelope) { a.content.WriteString(delta.Content) a.reasoning.WriteString(delta.ReasoningContent) + a.reasoning.WriteString(delta.Reasoning) for _, call := range delta.ToolCalls { if name := strings.TrimSpace(call.Function.Name); name != "" { a.toolCallNames = append(a.toolCallNames, name)