fix(openai): passthrough 응답 모델 alias를 보존한다
This commit is contained in:
toki
parent
8ff1489124
commit
909f916df8
3 changed files with 253 additions and 14 deletions
|
|
@ -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 검증 대상이 아니다.
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in a new issue