From 9d13ce8d032f2e00e328831a1fd77ef851a530a1 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 4 May 2026 10:58:47 +0000 Subject: [PATCH 01/13] fix(openai): support streaming image relay --- dto/openai_image.go | 12 ++--- dto/openai_image_test.go | 15 ++++++ relay/channel/openai/adaptor.go | 6 ++- relay/channel/openai/image_stream_test.go | 56 +++++++++++++++++++++++ relay/channel/openai/relay-openai.go | 56 +++++++++++++++++++++++ relay/helper/openai_image_request_test.go | 39 ++++++++++++++++ relay/helper/stream_scanner.go | 5 +- relay/helper/valid_request.go | 8 ++++ 8 files changed, 188 insertions(+), 9 deletions(-) create mode 100644 dto/openai_image_test.go create mode 100644 relay/channel/openai/image_stream_test.go create mode 100644 relay/helper/openai_image_request_test.go diff --git a/dto/openai_image.go b/dto/openai_image.go index fdef12b1a7df..416697e3b8cf 100644 --- a/dto/openai_image.go +++ b/dto/openai_image.go @@ -26,11 +26,11 @@ type ImageRequest struct { OutputFormat json.RawMessage `json:"output_format,omitempty"` OutputCompression json.RawMessage `json:"output_compression,omitempty"` PartialImages json.RawMessage `json:"partial_images,omitempty"` - // Stream bool `json:"stream,omitempty"` - Images json.RawMessage `json:"images,omitempty"` - Mask json.RawMessage `json:"mask,omitempty"` - InputFidelity json.RawMessage `json:"input_fidelity,omitempty"` - Watermark *bool `json:"watermark,omitempty"` + Stream bool `json:"stream,omitempty"` + Images json.RawMessage `json:"images,omitempty"` + Mask json.RawMessage `json:"mask,omitempty"` + InputFidelity json.RawMessage `json:"input_fidelity,omitempty"` + Watermark *bool `json:"watermark,omitempty"` // zhipu 4v WatermarkEnabled json.RawMessage `json:"watermark_enabled,omitempty"` UserId json.RawMessage `json:"user_id,omitempty"` @@ -163,7 +163,7 @@ func (i *ImageRequest) GetTokenCountMeta() *types.TokenCountMeta { } func (i *ImageRequest) IsStream(c *gin.Context) bool { - return false + return i.Stream } func (i *ImageRequest) SetModelName(modelName string) { diff --git a/dto/openai_image_test.go b/dto/openai_image_test.go new file mode 100644 index 000000000000..9ff1bc14dfc9 --- /dev/null +++ b/dto/openai_image_test.go @@ -0,0 +1,15 @@ +package dto + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestImageRequestStreamJSON(t *testing.T) { + var req ImageRequest + require.NoError(t, req.UnmarshalJSON([]byte(`{"model":"gpt-image-1","prompt":"draw a cat","stream":true}`))) + + require.True(t, req.Stream) + require.True(t, req.IsStream(nil)) +} diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index 6941ca54a732..94b5f3cf80a8 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -623,7 +623,11 @@ func (a *Adaptor) DoResponse(c *gin.Context, resp *http.Response, info *relaycom case relayconstant.RelayModeAudioTranscription: err, usage = OpenaiSTTHandler(c, resp, info, a.ResponseFormat) case relayconstant.RelayModeImagesGenerations, relayconstant.RelayModeImagesEdits: - usage, err = OpenaiHandlerWithUsage(c, info, resp) + if info.IsStream { + usage, err = OpenaiImageStreamHandler(c, info, resp) + } else { + usage, err = OpenaiHandlerWithUsage(c, info, resp) + } case relayconstant.RelayModeRerank: usage, err = common_handler.RerankHandler(c, info, resp) case relayconstant.RelayModeResponses: diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go new file mode 100644 index 000000000000..d5e9be16b609 --- /dev/null +++ b/relay/channel/openai/image_stream_test.go @@ -0,0 +1,56 @@ +package openai + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/QuantumNous/new-api/constant" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { + gin.SetMode(gin.TestMode) + + oldTimeout := constant.StreamingTimeout + constant.StreamingTimeout = 30 + t.Cleanup(func() { constant.StreamingTimeout = oldTimeout }) + + body := strings.Join([]string{ + `event: image_generation.partial_image`, + `data: {"type":"image_generation.partial_image","b64_json":"partial"}`, + ``, + `data: {"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7}}`, + ``, + `data: [DONE]`, + ``, + }, "\n") + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: true, + } + + usage, err := OpenaiImageStreamHandler(c, info, resp) + require.Nil(t, err) + require.Equal(t, 3, usage.PromptTokens) + require.Equal(t, 4, usage.CompletionTokens) + require.Equal(t, 7, usage.TotalTokens) + require.Contains(t, recorder.Body.String(), `event: image_generation.partial_image`) + require.Contains(t, recorder.Body.String(), `data: {"type":"image_generation.partial_image","b64_json":"partial"}`) + require.Contains(t, recorder.Body.String(), `data: {"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7}}`) + require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) +} diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index a85751844c0b..1f569bcef54a 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -1,6 +1,7 @@ package openai import ( + "bufio" "fmt" "io" "net/http" @@ -592,6 +593,61 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return &usageResp.Usage, nil } +func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + if resp == nil || resp.Body == nil { + logger.LogError(c, "invalid image stream response") + return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) + } + defer service.CloseResponseBodyGracefully(resp) + + usage := &dto.Usage{} + var lastStreamData []byte + + helper.SetEventStreamHeaders(c) + if info.StreamStatus == nil { + info.StreamStatus = relaycommon.NewStreamStatus() + } + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data:") { + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if data == "[DONE]" { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonDone, nil) + } else if data != "" { + info.SetFirstResponseTime() + info.ReceivedResponseCount++ + lastStreamData = common.StringToByteSlice(data) + var usageResp dto.SimpleResponse + if err := common.Unmarshal(lastStreamData, &usageResp); err == nil && service.ValidUsage(&usageResp.Usage) { + usage = &usageResp.Usage + } + } + } + if _, err := c.Writer.Write(append([]byte(line), '\n')); err != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + return usage, nil + } + if line == "" { + if err := helper.FlushWriter(c); err != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + return usage, nil + } + } + } + if err := scanner.Err(); err != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err) + } else if info.StreamStatus.EndReason == relaycommon.StreamEndReasonNone { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonEOF, nil) + } + _ = helper.FlushWriter(c) + + applyUsagePostProcessing(info, usage, lastStreamData) + return usage, nil +} + func applyUsagePostProcessing(info *relaycommon.RelayInfo, usage *dto.Usage, responseBody []byte) { if info == nil || usage == nil { return diff --git a/relay/helper/openai_image_request_test.go b/relay/helper/openai_image_request_test.go new file mode 100644 index 000000000000..5f2c2c94d1e8 --- /dev/null +++ b/relay/helper/openai_image_request_test.go @@ -0,0 +1,39 @@ +package helper + +import ( + "bytes" + "mime/multipart" + "net/http" + "net/http/httptest" + "testing" + + relayconstant "github.com/QuantumNous/new-api/relay/constant" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestGetAndValidOpenAIImageRequestMultipartStream(t *testing.T) { + gin.SetMode(gin.TestMode) + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + require.NoError(t, writer.WriteField("model", "gpt-image-1")) + require.NoError(t, writer.WriteField("prompt", "edit this image")) + require.NoError(t, writer.WriteField("stream", "true")) + require.NoError(t, writer.WriteField("n", "1")) + part, err := writer.CreateFormFile("image", "input.png") + require.NoError(t, err) + _, err = part.Write([]byte("fake image")) + require.NoError(t, err) + require.NoError(t, writer.Close()) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/edits", &body) + c.Request.Header.Set("Content-Type", writer.FormDataContentType()) + + req, err := GetAndValidOpenAIImageRequest(c, relayconstant.RelayModeImagesEdits) + require.NoError(t, err) + require.True(t, req.Stream) + require.True(t, req.IsStream(c)) +} diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index a9bc5e16a720..6c5323fc5f4a 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -40,8 +40,9 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon return } - // 无条件新建 StreamStatus - info.StreamStatus = relaycommon.NewStreamStatus() + if info.StreamStatus == nil { + info.StreamStatus = relaycommon.NewStreamStatus() + } // 确保响应体总是被关闭 defer func() { diff --git a/relay/helper/valid_request.go b/relay/helper/valid_request.go index 2581b2812c94..7b75d3bb891e 100644 --- a/relay/helper/valid_request.go +++ b/relay/helper/valid_request.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "math" + "strconv" "strings" "github.com/QuantumNous/new-api/common" @@ -154,6 +155,13 @@ func GetAndValidOpenAIImageRequest(c *gin.Context, relayMode int) (*dto.ImageReq imageRequest.N = common.GetPointer(uint(common.String2Int(formData.Get("n")))) imageRequest.Quality = formData.Get("quality") imageRequest.Size = formData.Get("size") + if streamValue := strings.TrimSpace(formData.Get("stream")); streamValue != "" { + stream, err := strconv.ParseBool(streamValue) + if err != nil { + return nil, fmt.Errorf("invalid stream value: %w", err) + } + imageRequest.Stream = stream + } if imageValue := formData.Get("image"); imageValue != "" { imageRequest.Image, _ = common.Marshal(imageValue) } From 9cc954225b786627255a2dc5804152c1e22773c3 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 4 May 2026 12:05:21 +0000 Subject: [PATCH 02/13] fix(openai): keep image edit multipart body reusable --- dto/openai_image_test.go | 1 + relay/channel/openai/adaptor.go | 10 +- relay/channel/openai/image_edit_test.go | 118 ++++++++++++++++++++++ relay/channel/openai/image_stream_test.go | 30 +++++- relay/channel/openai/relay-openai.go | 50 ++++++--- relay/helper/openai_image_request_test.go | 34 +++++++ relay/helper/stream_scanner.go | 5 +- relay/helper/valid_request.go | 7 +- 8 files changed, 230 insertions(+), 25 deletions(-) create mode 100644 relay/channel/openai/image_edit_test.go diff --git a/dto/openai_image_test.go b/dto/openai_image_test.go index 9ff1bc14dfc9..27e136377450 100644 --- a/dto/openai_image_test.go +++ b/dto/openai_image_test.go @@ -6,6 +6,7 @@ import ( "github.com/stretchr/testify/require" ) +// TestImageRequestStreamJSON verifies that image requests preserve stream=true. func TestImageRequestStreamJSON(t *testing.T) { var req ImageRequest require.NoError(t, req.UnmarshalJSON([]byte(`{"model":"gpt-image-1","prompt":"draw a cat","stream":true}`))) diff --git a/relay/channel/openai/adaptor.go b/relay/channel/openai/adaptor.go index 94b5f3cf80a8..fe797d98a2cf 100644 --- a/relay/channel/openai/adaptor.go +++ b/relay/channel/openai/adaptor.go @@ -9,6 +9,7 @@ import ( "mime/multipart" "net/http" "net/textproto" + "net/url" "path/filepath" "strings" @@ -437,10 +438,13 @@ func (a *Adaptor) ConvertImageRequest(c *gin.Context, info *relaycommon.RelayInf // 使用已解析的 multipart 表单，避免重复解析 mf := c.Request.MultipartForm if mf == nil { - if _, err := c.MultipartForm(); err != nil { - return nil, errors.New("failed to parse multipart form") + form, err := common.ParseMultipartFormReusable(c) + if err != nil { + return nil, fmt.Errorf("failed to parse multipart form: %w", err) } - mf = c.Request.MultipartForm + c.Request.MultipartForm = form + c.Request.PostForm = url.Values(form.Value) + mf = form } // 写入所有非文件字段 diff --git a/relay/channel/openai/image_edit_test.go b/relay/channel/openai/image_edit_test.go new file mode 100644 index 000000000000..051595842643 --- /dev/null +++ b/relay/channel/openai/image_edit_test.go @@ -0,0 +1,118 @@ +package openai + +import ( + "bytes" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "testing" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/dto" + relaycommon "github.com/QuantumNous/new-api/relay/common" + relayconstant "github.com/QuantumNous/new-api/relay/constant" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// TestConvertImageEditRequestKeepsValidMultipartStreamFields verifies multipart replay. +func TestConvertImageEditRequestKeepsValidMultipartStreamFields(t *testing.T) { + gin.SetMode(gin.TestMode) + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + require.NoError(t, writer.WriteField("model", "gpt-image-1")) + require.NoError(t, writer.WriteField("prompt", "edit this image")) + require.NoError(t, writer.WriteField("stream", "true")) + require.NoError(t, writer.WriteField("partial_images", "3")) + part, err := writer.CreateFormFile("image", "input.png") + require.NoError(t, err) + _, err = part.Write([]byte("fake image")) + require.NoError(t, err) + require.NoError(t, writer.Close()) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/edits", &body) + c.Request.Header.Set("Content-Type", writer.FormDataContentType()) + require.NoError(t, c.Request.ParseMultipartForm(32<<20)) + + info := &relaycommon.RelayInfo{ + RelayMode: relayconstant.RelayModeImagesEdits, + } + request := dto.ImageRequest{ + Model: "gpt-image-1", + Prompt: "edit this image", + Stream: true, + } + + converted, err := (&Adaptor{}).ConvertImageRequest(c, info, request) + require.NoError(t, err) + + convertedBody, ok := converted.(*bytes.Buffer) + require.True(t, ok) + + contentType := c.Request.Header.Get("Content-Type") + replayedRequest := httptest.NewRequest(http.MethodPost, "/v1/images/edits", bytes.NewReader(convertedBody.Bytes())) + replayedRequest.Header.Set("Content-Type", contentType) + require.NoError(t, replayedRequest.ParseMultipartForm(32<<20)) + + require.Equal(t, "gpt-image-1", replayedRequest.PostForm.Get("model")) + require.Equal(t, "edit this image", replayedRequest.PostForm.Get("prompt")) + require.Equal(t, "true", replayedRequest.PostForm.Get("stream")) + require.Equal(t, "3", replayedRequest.PostForm.Get("partial_images")) + require.Len(t, replayedRequest.MultipartForm.File["image"], 1) + + file, err := replayedRequest.MultipartForm.File["image"][0].Open() + require.NoError(t, err) + defer file.Close() + fileBytes, err := io.ReadAll(file) + require.NoError(t, err) + require.Equal(t, []byte("fake image"), fileBytes) +} + +// TestConvertImageEditRequestParsesReusableMultipartWhenFormIsMissing verifies fallback parsing. +func TestConvertImageEditRequestParsesReusableMultipartWhenFormIsMissing(t *testing.T) { + gin.SetMode(gin.TestMode) + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + require.NoError(t, writer.WriteField("model", "gpt-image-1")) + require.NoError(t, writer.WriteField("prompt", "edit without pre-parsed form")) + part, err := writer.CreateFormFile("image", "input.png") + require.NoError(t, err) + _, err = part.Write([]byte("fake image")) + require.NoError(t, err) + require.NoError(t, writer.Close()) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/edits", &body) + c.Request.Header.Set("Content-Type", writer.FormDataContentType()) + + storage, err := common.GetBodyStorage(c) + require.NoError(t, err) + c.Request.Body = io.NopCloser(storage) + c.Request.MultipartForm = nil + c.Request.PostForm = nil + + info := &relaycommon.RelayInfo{ + RelayMode: relayconstant.RelayModeImagesEdits, + } + request := dto.ImageRequest{ + Model: "gpt-image-1", + Prompt: "edit without pre-parsed form", + } + + converted, err := (&Adaptor{}).ConvertImageRequest(c, info, request) + require.NoError(t, err) + + convertedBody, ok := converted.(*bytes.Buffer) + require.True(t, ok) + replayedRequest := httptest.NewRequest(http.MethodPost, "/v1/images/edits", bytes.NewReader(convertedBody.Bytes())) + replayedRequest.Header.Set("Content-Type", c.Request.Header.Get("Content-Type")) + require.NoError(t, replayedRequest.ParseMultipartForm(32<<20)) + require.Equal(t, "edit without pre-parsed form", replayedRequest.PostForm.Get("prompt")) + require.Len(t, replayedRequest.MultipartForm.File["image"], 1) +} diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index d5e9be16b609..6052f4ccbfe8 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -8,13 +8,17 @@ import ( "testing" "github.com/QuantumNous/new-api/constant" + "github.com/QuantumNous/new-api/dto" relaycommon "github.com/QuantumNous/new-api/relay/common" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) +// TestOpenaiImageStreamHandlerForwardsSSEAndUsage verifies image SSE passthrough. func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { + oldMode := gin.Mode() gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) oldTimeout := constant.StreamingTimeout constant.StreamingTimeout = 30 @@ -24,7 +28,7 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { `event: image_generation.partial_image`, `data: {"type":"image_generation.partial_image","b64_json":"partial"}`, ``, - `data: {"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7}}`, + `data: {"usage":{"input_tokens":3,"output_tokens":4,"total_tokens":7,"input_tokens_details":{"image_tokens":2,"text_tokens":1}}}`, ``, `data: [DONE]`, ``, @@ -49,8 +53,30 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { require.Equal(t, 3, usage.PromptTokens) require.Equal(t, 4, usage.CompletionTokens) require.Equal(t, 7, usage.TotalTokens) + require.Equal(t, 2, usage.PromptTokensDetails.ImageTokens) + require.Equal(t, 1, usage.PromptTokensDetails.TextTokens) require.Contains(t, recorder.Body.String(), `event: image_generation.partial_image`) require.Contains(t, recorder.Body.String(), `data: {"type":"image_generation.partial_image","b64_json":"partial"}`) - require.Contains(t, recorder.Body.String(), `data: {"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7}}`) + require.Contains(t, recorder.Body.String(), `data: {"usage":{"input_tokens":3,"output_tokens":4,"total_tokens":7,"input_tokens_details":{"image_tokens":2,"text_tokens":1}}}`) require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) } + +// TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting verifies ImageRatio inputs. +func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testing.T) { + usage := &dto.Usage{ + InputTokens: 5000, + OutputTokens: 4000, + InputTokensDetails: &dto.InputTokenDetails{ + ImageTokens: 1000, + TextTokens: 4000, + }, + } + + normalizeOpenAIUsage(usage) + + require.Equal(t, 5000, usage.PromptTokens) + require.Equal(t, 4000, usage.CompletionTokens) + require.Equal(t, 9000, usage.TotalTokens) + require.Equal(t, 1000, usage.PromptTokensDetails.ImageTokens) + require.Equal(t, 4000, usage.PromptTokensDetails.TextTokens) +} diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 1f569bcef54a..94d80918d9f9 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -575,29 +575,44 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h // 写入新的 response body service.IOCopyBytesGracefully(c, resp, responseBody) - // Once we've written to the client, we should not return errors anymore - // because the upstream has already consumed resources and returned content - // We should still perform billing even if parsing fails - // format - if usageResp.InputTokens > 0 { - usageResp.PromptTokens += usageResp.InputTokens + normalizeOpenAIUsage(&usageResp.Usage) + applyUsagePostProcessing(info, &usageResp.Usage, responseBody) + return &usageResp.Usage, nil +} + +// normalizeOpenAIUsage maps OpenAI usage aliases into NewAPI billing fields. +func normalizeOpenAIUsage(usage *dto.Usage) { + if usage == nil { + return } - if usageResp.OutputTokens > 0 { - usageResp.CompletionTokens += usageResp.OutputTokens + if usage.InputTokens != 0 { + usage.PromptTokens = usage.InputTokens } - if usageResp.InputTokensDetails != nil { - usageResp.PromptTokensDetails.ImageTokens += usageResp.InputTokensDetails.ImageTokens - usageResp.PromptTokensDetails.TextTokens += usageResp.InputTokensDetails.TextTokens + if usage.OutputTokens != 0 { + usage.CompletionTokens = usage.OutputTokens + } + if usage.InputTokensDetails != nil { + usage.PromptTokensDetails.CachedTokens = usage.InputTokensDetails.CachedTokens + usage.PromptTokensDetails.ImageTokens = usage.InputTokensDetails.ImageTokens + usage.PromptTokensDetails.TextTokens = usage.InputTokensDetails.TextTokens + usage.PromptTokensDetails.AudioTokens = usage.InputTokensDetails.AudioTokens + } + if usage.TotalTokens == 0 { + usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens } - applyUsagePostProcessing(info, &usageResp.Usage, responseBody) - return &usageResp.Usage, nil } +// OpenaiImageStreamHandler forwards OpenAI Images SSE events and extracts usage. func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { if resp == nil || resp.Body == nil { logger.LogError(c, "invalid image stream response") return nil, types.NewOpenAIError(fmt.Errorf("invalid response"), types.ErrorCodeBadResponse, http.StatusInternalServerError) } + + contentType := strings.ToLower(resp.Header.Get("Content-Type")) + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices || !strings.Contains(contentType, "text/event-stream") { + return OpenaiHandlerWithUsage(c, info, resp) + } defer service.CloseResponseBodyGracefully(resp) usage := &dto.Usage{} @@ -609,7 +624,7 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } scanner := bufio.NewScanner(resp.Body) - scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.GetScannerBufferSize()) for scanner.Scan() { line := scanner.Text() if strings.HasPrefix(line, "data:") { @@ -621,8 +636,11 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp info.ReceivedResponseCount++ lastStreamData = common.StringToByteSlice(data) var usageResp dto.SimpleResponse - if err := common.Unmarshal(lastStreamData, &usageResp); err == nil && service.ValidUsage(&usageResp.Usage) { - usage = &usageResp.Usage + if err := common.Unmarshal(lastStreamData, &usageResp); err == nil { + normalizeOpenAIUsage(&usageResp.Usage) + if service.ValidUsage(&usageResp.Usage) { + usage = &usageResp.Usage + } } } } diff --git a/relay/helper/openai_image_request_test.go b/relay/helper/openai_image_request_test.go index 5f2c2c94d1e8..a0bb46c6aa41 100644 --- a/relay/helper/openai_image_request_test.go +++ b/relay/helper/openai_image_request_test.go @@ -2,16 +2,20 @@ package helper import ( "bytes" + "io" "mime/multipart" "net/http" "net/http/httptest" + "net/url" "testing" + "github.com/QuantumNous/new-api/common" relayconstant "github.com/QuantumNous/new-api/relay/constant" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) +// TestGetAndValidOpenAIImageRequestMultipartStream verifies reusable image edit parsing. func TestGetAndValidOpenAIImageRequestMultipartStream(t *testing.T) { gin.SetMode(gin.TestMode) @@ -26,6 +30,7 @@ func TestGetAndValidOpenAIImageRequestMultipartStream(t *testing.T) { _, err = part.Write([]byte("fake image")) require.NoError(t, err) require.NoError(t, writer.Close()) + originalBody := body.String() recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) @@ -36,4 +41,33 @@ func TestGetAndValidOpenAIImageRequestMultipartStream(t *testing.T) { require.NoError(t, err) require.True(t, req.Stream) require.True(t, req.IsStream(c)) + + bodyAfterValidation, err := io.ReadAll(c.Request.Body) + require.NoError(t, err) + require.Equal(t, originalBody, string(bodyAfterValidation)) + + form, err := common.ParseMultipartFormReusable(c) + require.NoError(t, err) + require.Equal(t, "true", url.Values(form.Value).Get("stream")) + require.Len(t, form.File["image"], 1) +} + +// TestGetAndValidOpenAIImageRequestMultipartStreamInvalidValue verifies stream validation. +func TestGetAndValidOpenAIImageRequestMultipartStreamInvalidValue(t *testing.T) { + gin.SetMode(gin.TestMode) + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + require.NoError(t, writer.WriteField("model", "gpt-image-1")) + require.NoError(t, writer.WriteField("stream", "notabool")) + require.NoError(t, writer.Close()) + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/edits", &body) + c.Request.Header.Set("Content-Type", writer.FormDataContentType()) + + _, err := GetAndValidOpenAIImageRequest(c, relayconstant.RelayModeImagesEdits) + require.Error(t, err) + require.Contains(t, err.Error(), "invalid stream value") } diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index 6c5323fc5f4a..df7aa39ea54a 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -27,7 +27,8 @@ const ( DefaultPingInterval = 10 * time.Second ) -func getScannerBufferSize() int { +// GetScannerBufferSize returns the configured maximum SSE scanner token size. +func GetScannerBufferSize() int { if constant.StreamScannerMaxBufferMB > 0 { return constant.StreamScannerMaxBufferMB << 20 } @@ -108,7 +109,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon close(stopChan) }() - scanner.Buffer(make([]byte, InitialScannerBufferSize), getScannerBufferSize()) + scanner.Buffer(make([]byte, InitialScannerBufferSize), GetScannerBufferSize()) scanner.Split(bufio.ScanLines) SetEventStreamHeaders(c) diff --git a/relay/helper/valid_request.go b/relay/helper/valid_request.go index 7b75d3bb891e..a53b63693d71 100644 --- a/relay/helper/valid_request.go +++ b/relay/helper/valid_request.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" "math" + "net/url" "strconv" "strings" @@ -145,11 +146,13 @@ func GetAndValidOpenAIImageRequest(c *gin.Context, relayMode int) (*dto.ImageReq switch relayMode { case relayconstant.RelayModeImagesEdits: if strings.Contains(c.Request.Header.Get("Content-Type"), "multipart/form-data") { - _, err := c.MultipartForm() + form, err := common.ParseMultipartFormReusable(c) if err != nil { return nil, fmt.Errorf("failed to parse image edit form request: %w", err) } - formData := c.Request.PostForm + formData := url.Values(form.Value) + c.Request.MultipartForm = form + c.Request.PostForm = formData imageRequest.Prompt = formData.Get("prompt") imageRequest.Model = formData.Get("model") imageRequest.N = common.GetPointer(uint(common.String2Int(formData.Get("n")))) From 47ba6eed3412ec8cdb2e67e493e41618b58bc04b Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 4 May 2026 14:53:45 +0000 Subject: [PATCH 03/13] test(openai): cover image stream usage details --- relay/channel/openai/image_stream_test.go | 7 +++++-- relay/channel/openai/relay-openai.go | 1 + 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index 6052f4ccbfe8..3dcfade620b4 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -58,6 +58,7 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { require.Contains(t, recorder.Body.String(), `event: image_generation.partial_image`) require.Contains(t, recorder.Body.String(), `data: {"type":"image_generation.partial_image","b64_json":"partial"}`) require.Contains(t, recorder.Body.String(), `data: {"usage":{"input_tokens":3,"output_tokens":4,"total_tokens":7,"input_tokens_details":{"image_tokens":2,"text_tokens":1}}}`) + require.Contains(t, recorder.Body.String(), `data: [DONE]`) require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) } @@ -67,8 +68,9 @@ func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testi InputTokens: 5000, OutputTokens: 4000, InputTokensDetails: &dto.InputTokenDetails{ - ImageTokens: 1000, - TextTokens: 4000, + CachedCreationTokens: 200, + ImageTokens: 1000, + TextTokens: 4000, }, } @@ -77,6 +79,7 @@ func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testi require.Equal(t, 5000, usage.PromptTokens) require.Equal(t, 4000, usage.CompletionTokens) require.Equal(t, 9000, usage.TotalTokens) + require.Equal(t, 200, usage.PromptTokensDetails.CachedCreationTokens) require.Equal(t, 1000, usage.PromptTokensDetails.ImageTokens) require.Equal(t, 4000, usage.PromptTokensDetails.TextTokens) } diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 94d80918d9f9..3cdfbb0a08ed 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -593,6 +593,7 @@ func normalizeOpenAIUsage(usage *dto.Usage) { } if usage.InputTokensDetails != nil { usage.PromptTokensDetails.CachedTokens = usage.InputTokensDetails.CachedTokens + usage.PromptTokensDetails.CachedCreationTokens = usage.InputTokensDetails.CachedCreationTokens usage.PromptTokensDetails.ImageTokens = usage.InputTokensDetails.ImageTokens usage.PromptTokensDetails.TextTokens = usage.InputTokensDetails.TextTokens usage.PromptTokensDetails.AudioTokens = usage.InputTokensDetails.AudioTokens From 71a66dcc5fd6fe948d9655b628823aba161bbb6a Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 5 May 2026 09:24:51 +0000 Subject: [PATCH 04/13] test(openai): cover image edit fallback stream field --- relay/channel/openai/image_edit_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/relay/channel/openai/image_edit_test.go b/relay/channel/openai/image_edit_test.go index 051595842643..b37551b04da7 100644 --- a/relay/channel/openai/image_edit_test.go +++ b/relay/channel/openai/image_edit_test.go @@ -80,6 +80,7 @@ func TestConvertImageEditRequestParsesReusableMultipartWhenFormIsMissing(t *test writer := multipart.NewWriter(&body) require.NoError(t, writer.WriteField("model", "gpt-image-1")) require.NoError(t, writer.WriteField("prompt", "edit without pre-parsed form")) + require.NoError(t, writer.WriteField("stream", "true")) part, err := writer.CreateFormFile("image", "input.png") require.NoError(t, err) _, err = part.Write([]byte("fake image")) @@ -103,6 +104,7 @@ func TestConvertImageEditRequestParsesReusableMultipartWhenFormIsMissing(t *test request := dto.ImageRequest{ Model: "gpt-image-1", Prompt: "edit without pre-parsed form", + Stream: true, } converted, err := (&Adaptor{}).ConvertImageRequest(c, info, request) @@ -114,5 +116,6 @@ func TestConvertImageEditRequestParsesReusableMultipartWhenFormIsMissing(t *test replayedRequest.Header.Set("Content-Type", c.Request.Header.Get("Content-Type")) require.NoError(t, replayedRequest.ParseMultipartForm(32<<20)) require.Equal(t, "edit without pre-parsed form", replayedRequest.PostForm.Get("prompt")) + require.Equal(t, "true", replayedRequest.PostForm.Get("stream")) require.Len(t, replayedRequest.MultipartForm.File["image"], 1) } From e79107aabf92d699b1a5fc2c05c2ea3f8ac14464 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 10 May 2026 19:34:29 +0000 Subject: [PATCH 05/13] fix(openai): wrap image json fallback as stream --- relay/channel/openai/image_stream_test.go | 39 ++++++++++ relay/channel/openai/relay-openai.go | 95 ++++++++++++++++++++++- 2 files changed, 133 insertions(+), 1 deletion(-) diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index 3dcfade620b4..651348eb61e8 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -62,6 +62,45 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) } +// TestOpenaiImageStreamHandlerWrapsJSONResponse verifies stream requests keep SSE semantics +// even when the upstream image endpoint ignores stream=true and returns a JSON body. +func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { + oldMode := gin.Mode() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) + + body := `{"created":1710000000,"data":[{"b64_json":"final","revised_prompt":"draw a cat"}],"usage":{"input_tokens":3,"output_tokens":4,"total_tokens":7,"input_tokens_details":{"image_tokens":2,"text_tokens":1}}}` + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"application/json"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: true, + } + + usage, err := OpenaiImageStreamHandler(c, info, resp) + require.Nil(t, err) + require.Equal(t, 3, usage.PromptTokens) + require.Equal(t, 4, usage.CompletionTokens) + require.Equal(t, 7, usage.TotalTokens) + require.Equal(t, 2, usage.PromptTokensDetails.ImageTokens) + require.Equal(t, 1, usage.PromptTokensDetails.TextTokens) + require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) + require.Empty(t, recorder.Header().Get("Content-Length")) + require.Contains(t, recorder.Body.String(), `event: image_generation.completed`) + require.Contains(t, recorder.Body.String(), `"type":"image_generation.completed"`) + require.Contains(t, recorder.Body.String(), `"b64_json":"final"`) + require.Contains(t, recorder.Body.String(), `"revised_prompt":"draw a cat"`) + require.Contains(t, recorder.Body.String(), `data: [DONE]`) +} + // TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting verifies ImageRatio inputs. func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testing.T) { usage := &dto.Usage{ diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 3cdfbb0a08ed..dd247064a459 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "strings" + "time" "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/constant" @@ -611,9 +612,12 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } contentType := strings.ToLower(resp.Header.Get("Content-Type")) - if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices || !strings.Contains(contentType, "text/event-stream") { + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { return OpenaiHandlerWithUsage(c, info, resp) } + if !strings.Contains(contentType, "text/event-stream") { + return OpenaiImageJSONAsStreamHandler(c, info, resp) + } defer service.CloseResponseBodyGracefully(resp) usage := &dto.Usage{} @@ -667,6 +671,95 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } +func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { + defer service.CloseResponseBodyGracefully(resp) + + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError) + } + + var imageResp dto.ImageResponse + if err := common.Unmarshal(responseBody, &imageResp); err != nil { + return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) + } + + var usageResp dto.SimpleResponse + _ = common.Unmarshal(responseBody, &usageResp) + normalizeOpenAIUsage(&usageResp.Usage) + applyUsagePostProcessing(info, &usageResp.Usage, responseBody) + + helper.SetEventStreamHeaders(c) + c.Status(http.StatusOK) + + created := imageResp.Created + if created == 0 { + created = time.Now().Unix() + } + for _, image := range imageResp.Data { + payload := map[string]any{ + "type": "image_generation.completed", + "created_at": created, + } + if image.Url != "" { + payload["url"] = image.Url + } + if image.B64Json != "" { + payload["b64_json"] = image.B64Json + } + if image.RevisedPrompt != "" { + payload["revised_prompt"] = image.RevisedPrompt + } + if service.ValidUsage(&usageResp.Usage) { + payload["usage"] = usageResp.Usage + } + if err := writeOpenaiImageStreamPayload(c, "image_generation.completed", payload); err != nil { + if info != nil && info.StreamStatus != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + } + return &usageResp.Usage, nil + } + } + if err := writeOpenaiImageStreamDone(c); err != nil { + if info != nil && info.StreamStatus != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + } + return &usageResp.Usage, nil + } + if info != nil { + info.SetFirstResponseTime() + info.ReceivedResponseCount += len(imageResp.Data) + if info.StreamStatus == nil { + info.StreamStatus = relaycommon.NewStreamStatus() + } + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonDone, nil) + } + return &usageResp.Usage, nil +} + +func writeOpenaiImageStreamPayload(c *gin.Context, eventName string, payload any) error { + data, err := common.Marshal(payload) + if err != nil { + return err + } + if eventName != "" { + if _, err := fmt.Fprintf(c.Writer, "event: %s\n", eventName); err != nil { + return err + } + } + if _, err := fmt.Fprintf(c.Writer, "data: %s\n\n", data); err != nil { + return err + } + return helper.FlushWriter(c) +} + +func writeOpenaiImageStreamDone(c *gin.Context) error { + if _, err := fmt.Fprint(c.Writer, "data: [DONE]\n\n"); err != nil { + return err + } + return helper.FlushWriter(c) +} + func applyUsagePostProcessing(info *relaycommon.RelayInfo, usage *dto.Usage, responseBody []byte) { if info == nil || usage == nil { return From 9a9609d7e73cf8a2c7ebb6344ec282d5f90bfccd Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 11 May 2026 21:48:52 +0000 Subject: [PATCH 06/13] fix(relay): support OpenAI image streaming --- relay/channel/openai/image_stream_test.go | 4 --- relay/channel/openai/relay-openai.go | 34 ++++++++++++++--------- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index 651348eb61e8..6540b49fa912 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -14,7 +14,6 @@ import ( "github.com/stretchr/testify/require" ) -// TestOpenaiImageStreamHandlerForwardsSSEAndUsage verifies image SSE passthrough. func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { oldMode := gin.Mode() gin.SetMode(gin.TestMode) @@ -62,8 +61,6 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) } -// TestOpenaiImageStreamHandlerWrapsJSONResponse verifies stream requests keep SSE semantics -// even when the upstream image endpoint ignores stream=true and returns a JSON body. func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { oldMode := gin.Mode() gin.SetMode(gin.TestMode) @@ -101,7 +98,6 @@ func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { require.Contains(t, recorder.Body.String(), `data: [DONE]`) } -// TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting verifies ImageRatio inputs. func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testing.T) { usage := &dto.Usage{ InputTokens: 5000, diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index dd247064a459..79b719324aa3 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -581,7 +581,6 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return &usageResp.Usage, nil } -// normalizeOpenAIUsage maps OpenAI usage aliases into NewAPI billing fields. func normalizeOpenAIUsage(usage *dto.Usage) { if usage == nil { return @@ -604,7 +603,6 @@ func normalizeOpenAIUsage(usage *dto.Usage) { } } -// OpenaiImageStreamHandler forwards OpenAI Images SSE events and extracts usage. func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { if resp == nil || resp.Body == nil { logger.LogError(c, "invalid image stream response") @@ -624,21 +622,25 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp var lastStreamData []byte helper.SetEventStreamHeaders(c) - if info.StreamStatus == nil { + if info != nil && info.StreamStatus == nil { info.StreamStatus = relaycommon.NewStreamStatus() } scanner := bufio.NewScanner(resp.Body) - scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.GetScannerBufferSize()) + scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) for scanner.Scan() { line := scanner.Text() if strings.HasPrefix(line, "data:") { data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) if data == "[DONE]" { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonDone, nil) + if info != nil && info.StreamStatus != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonDone, nil) + } } else if data != "" { - info.SetFirstResponseTime() - info.ReceivedResponseCount++ + if info != nil { + info.SetFirstResponseTime() + info.ReceivedResponseCount++ + } lastStreamData = common.StringToByteSlice(data) var usageResp dto.SimpleResponse if err := common.Unmarshal(lastStreamData, &usageResp); err == nil { @@ -650,20 +652,26 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } } if _, err := c.Writer.Write(append([]byte(line), '\n')); err != nil { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + if info != nil && info.StreamStatus != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + } return usage, nil } if line == "" { if err := helper.FlushWriter(c); err != nil { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + if info != nil && info.StreamStatus != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonClientGone, err) + } return usage, nil } } } - if err := scanner.Err(); err != nil { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err) - } else if info.StreamStatus.EndReason == relaycommon.StreamEndReasonNone { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonEOF, nil) + if info != nil && info.StreamStatus != nil { + if err := scanner.Err(); err != nil { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err) + } else if info.StreamStatus.EndReason == relaycommon.StreamEndReasonNone { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonEOF, nil) + } } _ = helper.FlushWriter(c) From 0fb42bece9fd2a622032cbf41549db6f48f3a382 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 12 May 2026 02:22:52 +0000 Subject: [PATCH 07/13] fix(openai): record image stream upstream error events --- relay/channel/openai/image_stream_test.go | 40 +++++++++++++++ relay/channel/openai/relay-openai.go | 60 ++++++++++++++++++++++- 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index 6540b49fa912..b843d7a905a9 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -98,6 +98,46 @@ func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { require.Contains(t, recorder.Body.String(), `data: [DONE]`) } +func TestOpenaiImageStreamHandlerRecordsUpstreamErrorEvent(t *testing.T) { + oldMode := gin.Mode() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) + + body := strings.Join([]string{ + `event: image_generation.partial_image`, + `data: {"type":"image_generation.partial_image","b64_json":"partial"}`, + ``, + `event: error`, + `data: {"type":"upstream_error","error":{"message":"stream error: stream ID 77; INTERNAL_ERROR; received from peer"}}`, + ``, + }, "\n") + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: true, + } + + usage, err := OpenaiImageStreamHandler(c, info, resp) + require.Nil(t, err) + require.NotNil(t, usage) + require.NotNil(t, info.StreamStatus) + require.Equal(t, relaycommon.StreamEndReasonHandlerStop, info.StreamStatus.EndReason) + require.True(t, info.StreamStatus.HasErrors()) + require.Equal(t, 1, info.StreamStatus.TotalErrorCount()) + require.Contains(t, info.StreamStatus.Errors[0].Message, "INTERNAL_ERROR") + require.Contains(t, recorder.Body.String(), `event: error`) + require.Contains(t, recorder.Body.String(), `stream ID 77`) +} + func TestNormalizeOpenAIUsageMapsImageTokenDetailsWithoutDoubleCounting(t *testing.T) { usage := &dto.Usage{ InputTokens: 5000, diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 79b719324aa3..3f304c363cc6 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -2,6 +2,7 @@ package openai import ( "bufio" + "encoding/json" "fmt" "io" "net/http" @@ -628,9 +629,12 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp scanner := bufio.NewScanner(resp.Body) scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + currentEvent := "" for scanner.Scan() { line := scanner.Text() - if strings.HasPrefix(line, "data:") { + if strings.HasPrefix(line, "event:") { + currentEvent = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + } else if strings.HasPrefix(line, "data:") { data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) if data == "[DONE]" { if info != nil && info.StreamStatus != nil { @@ -642,6 +646,9 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp info.ReceivedResponseCount++ } lastStreamData = common.StringToByteSlice(data) + if info != nil && info.StreamStatus != nil && isOpenAIImageStreamErrorEvent(currentEvent, lastStreamData) { + info.StreamStatus.RecordError(extractOpenAIImageStreamErrorMessage(lastStreamData)) + } var usageResp dto.SimpleResponse if err := common.Unmarshal(lastStreamData, &usageResp); err == nil { normalizeOpenAIUsage(&usageResp.Usage) @@ -664,11 +671,14 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } return usage, nil } + currentEvent = "" } } if info != nil && info.StreamStatus != nil { if err := scanner.Err(); err != nil { info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err) + } else if info.StreamStatus.HasErrors() { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonHandlerStop, fmt.Errorf("upstream image stream returned error event")) } else if info.StreamStatus.EndReason == relaycommon.StreamEndReasonNone { info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonEOF, nil) } @@ -679,6 +689,54 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } +func isOpenAIImageStreamErrorEvent(eventName string, data []byte) bool { + if strings.EqualFold(strings.TrimSpace(eventName), "error") { + return true + } + if !json.Valid(data) { + return false + } + var payload struct { + Type string `json:"type"` + Error json.RawMessage `json:"error"` + } + if err := common.Unmarshal(data, &payload); err != nil { + return false + } + payloadType := strings.ToLower(strings.TrimSpace(payload.Type)) + return payloadType == "error" || payloadType == "upstream_error" || len(payload.Error) > 0 +} + +func extractOpenAIImageStreamErrorMessage(data []byte) string { + if len(data) == 0 || !json.Valid(data) { + return "upstream image stream returned error event" + } + var payload struct { + Message string `json:"message"` + Error json.RawMessage `json:"error"` + } + if err := common.Unmarshal(data, &payload); err != nil { + return "upstream image stream returned error event" + } + if msg := strings.TrimSpace(payload.Message); msg != "" { + return msg + } + if len(payload.Error) > 0 { + var nested struct { + Message string `json:"message"` + } + if err := common.Unmarshal(payload.Error, &nested); err == nil { + if msg := strings.TrimSpace(nested.Message); msg != "" { + return msg + } + } + if msg := strings.TrimSpace(common.JsonRawMessageToString(payload.Error)); msg != "" { + return msg + } + } + return "upstream image stream returned error event" +} + func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { defer service.CloseResponseBodyGracefully(resp) From 28179640d47eea11877df52aec4f9b2b4ae6a4da Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 12 May 2026 04:21:36 +0000 Subject: [PATCH 08/13] fix(openai): harden image stream relay --- relay/channel/openai/image_stream_test.go | 31 +++++++++++++++++++++++ relay/channel/openai/relay-openai.go | 27 +++++++++++++++----- 2 files changed, 51 insertions(+), 7 deletions(-) diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index b843d7a905a9..23c9f7b5aef0 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -10,6 +10,7 @@ import ( "github.com/QuantumNous/new-api/constant" "github.com/QuantumNous/new-api/dto" relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/relay/helper" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) @@ -61,6 +62,36 @@ func TestOpenaiImageStreamHandlerForwardsSSEAndUsage(t *testing.T) { require.Equal(t, "text/event-stream", recorder.Header().Get("Content-Type")) } +func TestOpenaiImageStreamHandlerForwardsLargeSSELine(t *testing.T) { + oldMode := gin.Mode() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) + + payload := strings.Repeat("x", helper.DefaultMaxScannerBufferSize+1) + body := "data: " + payload + "\n\n" + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: true, + } + + usage, err := OpenaiImageStreamHandler(c, info, resp) + require.Nil(t, err) + require.NotNil(t, usage) + require.Contains(t, recorder.Body.String(), payload) + require.NotNil(t, info.StreamStatus) + require.Equal(t, relaycommon.StreamEndReasonEOF, info.StreamStatus.EndReason) +} + func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { oldMode := gin.Mode() gin.SetMode(gin.TestMode) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 3f304c363cc6..af37155684c4 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -627,11 +627,19 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp info.StreamStatus = relaycommon.NewStreamStatus() } - scanner := bufio.NewScanner(resp.Body) - scanner.Buffer(make([]byte, helper.InitialScannerBufferSize), helper.DefaultMaxScannerBufferSize) + reader := bufio.NewReader(resp.Body) currentEvent := "" - for scanner.Scan() { - line := scanner.Text() + var readErr error + for { + line, err := reader.ReadString('\n') + if err != nil { + readErr = err + if len(line) == 0 { + break + } + } + line = strings.TrimSuffix(line, "\n") + line = strings.TrimSuffix(line, "\r") if strings.HasPrefix(line, "event:") { currentEvent = strings.TrimSpace(strings.TrimPrefix(line, "event:")) } else if strings.HasPrefix(line, "data:") { @@ -673,10 +681,13 @@ func OpenaiImageStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } currentEvent = "" } + if readErr != nil { + break + } } if info != nil && info.StreamStatus != nil { - if err := scanner.Err(); err != nil { - info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, err) + if readErr != nil && readErr != io.EOF { + info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonScannerErr, readErr) } else if info.StreamStatus.HasErrors() { info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonHandlerStop, fmt.Errorf("upstream image stream returned error event")) } else if info.StreamStatus.EndReason == relaycommon.StreamEndReasonNone { @@ -762,6 +773,9 @@ func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, if created == 0 { created = time.Now().Unix() } + if info != nil { + info.SetFirstResponseTime() + } for _, image := range imageResp.Data { payload := map[string]any{ "type": "image_generation.completed", @@ -793,7 +807,6 @@ func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, return &usageResp.Usage, nil } if info != nil { - info.SetFirstResponseTime() info.ReceivedResponseCount += len(imageResp.Data) if info.StreamStatus == nil { info.StreamStatus = relaycommon.NewStreamStatus() From 5d8d6c20591a5d8a248613c307f7f308ee829425 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 17 May 2026 14:51:22 +0000 Subject: [PATCH 09/13] fix(openai): return image JSON errors --- dto/openai_response.go | 29 ++++++++++ relay/channel/openai/image_stream_test.go | 67 +++++++++++++++++++++++ relay/channel/openai/relay-openai.go | 17 ++++++ types/error.go | 13 +++-- 4 files changed, 121 insertions(+), 5 deletions(-) diff --git a/dto/openai_response.go b/dto/openai_response.go index 0e6b818dbd8b..6308f20de1b0 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -429,6 +429,35 @@ func GetOpenAIError(errorField any) *types.OpenAIError { if errCode, ok := err["code"]; ok { openaiErr.Code = errCode } + if errStatus, ok := err["status"]; ok { + switch status := errStatus.(type) { + case int: + openaiErr.Status = status + case float64: + openaiErr.Status = int(status) + case json.Number: + if value, parseErr := status.Int64(); parseErr == nil { + openaiErr.Status = int(value) + } + } + } + if generationID, ok := err["generation_id"].(string); ok { + openaiErr.GenerationID = generationID + } + if creditsConsumed, ok := err["credits_consumed"]; ok { + switch credits := creditsConsumed.(type) { + case int: + openaiErr.CreditsConsumed = &credits + case float64: + value := int(credits) + openaiErr.CreditsConsumed = &value + case json.Number: + if value, parseErr := credits.Int64(); parseErr == nil { + consumed := int(value) + openaiErr.CreditsConsumed = &consumed + } + } + } return openaiErr case string: // 处理简单字符串错误 diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index 23c9f7b5aef0..d5e22dc949a9 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -129,6 +129,73 @@ func TestOpenaiImageStreamHandlerWrapsJSONResponse(t *testing.T) { require.Contains(t, recorder.Body.String(), `data: [DONE]`) } +func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { + oldMode := gin.Mode() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) + + body := `{"error":{"message":"content moderation failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_123","credits_consumed":0}}` + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"application/json"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: false, + } + + usage, err := OpenaiHandlerWithUsage(c, info, resp) + require.Nil(t, usage) + require.NotNil(t, err) + require.Equal(t, http.StatusBadGateway, err.StatusCode) + oaiError := err.ToOpenAIError() + require.Equal(t, "content moderation failed", oaiError.Message) + require.Equal(t, "upstream_error", oaiError.Type) + require.Equal(t, "content_moderation_failed", oaiError.Code) + require.Equal(t, 502, oaiError.Status) + require.Equal(t, "gen_123", oaiError.GenerationID) + require.NotNil(t, oaiError.CreditsConsumed) + require.Equal(t, 0, *oaiError.CreditsConsumed) + require.Empty(t, recorder.Body.String()) +} + +func TestOpenaiImageStreamHandlerReturnsJSONErrorFallback(t *testing.T) { + oldMode := gin.Mode() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(oldMode) }) + + body := `{"error":{"message":"image edit failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_stream","credits_consumed":0}}` + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/images/generations", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(body)), + Header: http.Header{"Content-Type": []string{"application/json"}}, + } + info := &relaycommon.RelayInfo{ + ChannelMeta: &relaycommon.ChannelMeta{}, + IsStream: true, + } + + usage, err := OpenaiImageStreamHandler(c, info, resp) + require.Nil(t, usage) + require.NotNil(t, err) + require.Equal(t, http.StatusBadGateway, err.StatusCode) + oaiError := err.ToOpenAIError() + require.Equal(t, "image edit failed", oaiError.Message) + require.Equal(t, "gen_stream", oaiError.GenerationID) + require.Empty(t, recorder.Body.String()) +} + func TestOpenaiImageStreamHandlerRecordsUpstreamErrorEvent(t *testing.T) { oldMode := gin.Mode() gin.SetMode(gin.TestMode) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index af37155684c4..4d3957a406d6 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -574,6 +574,10 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) } + if oaiError := usageResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { + return nil, types.WithOpenAIError(*oaiError, openAIErrorStatusCode(oaiError, resp.StatusCode)) + } + // 写入新的 response body service.IOCopyBytesGracefully(c, resp, responseBody) @@ -582,6 +586,16 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return &usageResp.Usage, nil } +func openAIErrorStatusCode(oaiError *types.OpenAIError, fallback int) int { + if oaiError != nil && oaiError.Status >= http.StatusBadRequest && oaiError.Status <= 599 { + return oaiError.Status + } + if fallback >= http.StatusBadRequest && fallback <= 599 { + return fallback + } + return http.StatusBadGateway +} + func normalizeOpenAIUsage(usage *dto.Usage) { if usage == nil { return @@ -763,6 +777,9 @@ func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, var usageResp dto.SimpleResponse _ = common.Unmarshal(responseBody, &usageResp) + if oaiError := usageResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { + return nil, types.WithOpenAIError(*oaiError, openAIErrorStatusCode(oaiError, resp.StatusCode)) + } normalizeOpenAIUsage(&usageResp.Usage) applyUsagePostProcessing(info, &usageResp.Usage, responseBody) diff --git a/types/error.go b/types/error.go index 9717401ae7b2..b6aa8fa03535 100644 --- a/types/error.go +++ b/types/error.go @@ -11,11 +11,14 @@ import ( ) type OpenAIError struct { - Message string `json:"message"` - Type string `json:"type"` - Param string `json:"param"` - Code any `json:"code"` - Metadata json.RawMessage `json:"metadata,omitempty"` + Message string `json:"message"` + Type string `json:"type"` + Param string `json:"param"` + Code any `json:"code"` + Status int `json:"status,omitempty"` + GenerationID string `json:"generation_id,omitempty"` + CreditsConsumed *int `json:"credits_consumed,omitempty"` + Metadata json.RawMessage `json:"metadata,omitempty"` } type ClaudeError struct { From 01dd22d0d3aa4d103049a589fab43aabd8bc5e96 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 29 May 2026 05:01:42 +0000 Subject: [PATCH 10/13] fix(relay): reset stream status per scanner run --- relay/helper/stream_scanner.go | 4 +--- relay/helper/stream_scanner_test.go | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index 27c5ea9d4182..9c24a893f6e0 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -41,9 +41,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon return } - if info.StreamStatus == nil { - info.StreamStatus = relaycommon.NewStreamStatus() - } + info.StreamStatus = relaycommon.NewStreamStatus() // 确保响应体总是被关闭 defer func() { diff --git a/relay/helper/stream_scanner_test.go b/relay/helper/stream_scanner_test.go index 9d6f3bb49123..e31766a78862 100644 --- a/relay/helper/stream_scanner_test.go +++ b/relay/helper/stream_scanner_test.go @@ -614,7 +614,7 @@ func TestStreamScannerHandler_StreamStatus_InitializedIfNil(t *testing.T) { assert.NotNil(t, info.StreamStatus) } -func TestStreamScannerHandler_StreamStatus_PreInitialized(t *testing.T) { +func TestStreamScannerHandler_StreamStatus_ReplacesPreInitialized(t *testing.T) { t.Parallel() body := buildSSEBody(5) @@ -626,7 +626,7 @@ func TestStreamScannerHandler_StreamStatus_PreInitialized(t *testing.T) { StreamScannerHandler(c, resp, info, func(data string, sr *StreamResult) {}) assert.Equal(t, relaycommon.StreamEndReasonDone, info.StreamStatus.EndReason) - assert.Equal(t, 1, info.StreamStatus.TotalErrorCount()) + assert.Equal(t, 0, info.StreamStatus.TotalErrorCount()) } func TestStreamScannerHandler_PingInterleavesWithSlowUpstream(t *testing.T) { From d8a008881e1f9b1fb1ca96b53568c69a990dae50 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 29 May 2026 05:22:58 +0000 Subject: [PATCH 11/13] fix(relay): drop upstream credit passthrough --- dto/openai_response.go | 14 -------------- relay/channel/openai/image_stream_test.go | 6 ++---- types/error.go | 15 +++++++-------- 3 files changed, 9 insertions(+), 26 deletions(-) diff --git a/dto/openai_response.go b/dto/openai_response.go index 6308f20de1b0..4ee8424bd35b 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -444,20 +444,6 @@ func GetOpenAIError(errorField any) *types.OpenAIError { if generationID, ok := err["generation_id"].(string); ok { openaiErr.GenerationID = generationID } - if creditsConsumed, ok := err["credits_consumed"]; ok { - switch credits := creditsConsumed.(type) { - case int: - openaiErr.CreditsConsumed = &credits - case float64: - value := int(credits) - openaiErr.CreditsConsumed = &value - case json.Number: - if value, parseErr := credits.Int64(); parseErr == nil { - consumed := int(value) - openaiErr.CreditsConsumed = &consumed - } - } - } return openaiErr case string: // 处理简单字符串错误 diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index d5e22dc949a9..d76e4de715cc 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -134,7 +134,7 @@ func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { gin.SetMode(gin.TestMode) t.Cleanup(func() { gin.SetMode(oldMode) }) - body := `{"error":{"message":"content moderation failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_123","credits_consumed":0}}` + body := `{"error":{"message":"content moderation failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_123"}}` recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) @@ -160,8 +160,6 @@ func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { require.Equal(t, "content_moderation_failed", oaiError.Code) require.Equal(t, 502, oaiError.Status) require.Equal(t, "gen_123", oaiError.GenerationID) - require.NotNil(t, oaiError.CreditsConsumed) - require.Equal(t, 0, *oaiError.CreditsConsumed) require.Empty(t, recorder.Body.String()) } @@ -170,7 +168,7 @@ func TestOpenaiImageStreamHandlerReturnsJSONErrorFallback(t *testing.T) { gin.SetMode(gin.TestMode) t.Cleanup(func() { gin.SetMode(oldMode) }) - body := `{"error":{"message":"image edit failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_stream","credits_consumed":0}}` + body := `{"error":{"message":"image edit failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_stream"}}` recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) diff --git a/types/error.go b/types/error.go index b6aa8fa03535..2204c7d1aa98 100644 --- a/types/error.go +++ b/types/error.go @@ -11,14 +11,13 @@ import ( ) type OpenAIError struct { - Message string `json:"message"` - Type string `json:"type"` - Param string `json:"param"` - Code any `json:"code"` - Status int `json:"status,omitempty"` - GenerationID string `json:"generation_id,omitempty"` - CreditsConsumed *int `json:"credits_consumed,omitempty"` - Metadata json.RawMessage `json:"metadata,omitempty"` + Message string `json:"message"` + Type string `json:"type"` + Param string `json:"param"` + Code any `json:"code"` + Status int `json:"status,omitempty"` + GenerationID string `json:"generation_id,omitempty"` + Metadata json.RawMessage `json:"metadata,omitempty"` } type ClaudeError struct { From 518f804d61fd4126138c2493233424a6e0caa2cc Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 29 May 2026 09:45:27 +0000 Subject: [PATCH 12/13] fix(openai): keep image errors minimal --- dto/openai_response.go | 3 --- relay/channel/openai/image_stream_test.go | 6 ++---- types/error.go | 1 - 3 files changed, 2 insertions(+), 8 deletions(-) diff --git a/dto/openai_response.go b/dto/openai_response.go index 4ee8424bd35b..7f696849df5f 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -441,9 +441,6 @@ func GetOpenAIError(errorField any) *types.OpenAIError { } } } - if generationID, ok := err["generation_id"].(string); ok { - openaiErr.GenerationID = generationID - } return openaiErr case string: // 处理简单字符串错误 diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index d76e4de715cc..c72c15169e7f 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -134,7 +134,7 @@ func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { gin.SetMode(gin.TestMode) t.Cleanup(func() { gin.SetMode(oldMode) }) - body := `{"error":{"message":"content moderation failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_123"}}` + body := `{"error":{"message":"content moderation failed","type":"upstream_error","code":"content_moderation_failed","status":502}}` recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) @@ -159,7 +159,6 @@ func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { require.Equal(t, "upstream_error", oaiError.Type) require.Equal(t, "content_moderation_failed", oaiError.Code) require.Equal(t, 502, oaiError.Status) - require.Equal(t, "gen_123", oaiError.GenerationID) require.Empty(t, recorder.Body.String()) } @@ -168,7 +167,7 @@ func TestOpenaiImageStreamHandlerReturnsJSONErrorFallback(t *testing.T) { gin.SetMode(gin.TestMode) t.Cleanup(func() { gin.SetMode(oldMode) }) - body := `{"error":{"message":"image edit failed","type":"upstream_error","code":"content_moderation_failed","status":502,"generation_id":"gen_stream"}}` + body := `{"error":{"message":"image edit failed","type":"upstream_error","code":"content_moderation_failed","status":502}}` recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) @@ -190,7 +189,6 @@ func TestOpenaiImageStreamHandlerReturnsJSONErrorFallback(t *testing.T) { require.Equal(t, http.StatusBadGateway, err.StatusCode) oaiError := err.ToOpenAIError() require.Equal(t, "image edit failed", oaiError.Message) - require.Equal(t, "gen_stream", oaiError.GenerationID) require.Empty(t, recorder.Body.String()) } diff --git a/types/error.go b/types/error.go index 2204c7d1aa98..e4c0a1380ac1 100644 --- a/types/error.go +++ b/types/error.go @@ -16,7 +16,6 @@ type OpenAIError struct { Param string `json:"param"` Code any `json:"code"` Status int `json:"status,omitempty"` - GenerationID string `json:"generation_id,omitempty"` Metadata json.RawMessage `json:"metadata,omitempty"` } From 5284fe7f9eb92ce48ddc1cf158a371758778e991 Mon Sep 17 00:00:00 2001 From: CaIon Date: Sun, 31 May 2026 14:14:28 +0800 Subject: [PATCH 13/13] fix(openai): keep image error status from response --- dto/openai_response.go | 12 ------------ relay/channel/openai/image_stream_test.go | 5 ++--- relay/channel/openai/relay-openai.go | 14 ++------------ relay/helper/stream_scanner.go | 6 +++--- types/error.go | 11 +++++------ 5 files changed, 12 insertions(+), 36 deletions(-) diff --git a/dto/openai_response.go b/dto/openai_response.go index 7f696849df5f..0e6b818dbd8b 100644 --- a/dto/openai_response.go +++ b/dto/openai_response.go @@ -429,18 +429,6 @@ func GetOpenAIError(errorField any) *types.OpenAIError { if errCode, ok := err["code"]; ok { openaiErr.Code = errCode } - if errStatus, ok := err["status"]; ok { - switch status := errStatus.(type) { - case int: - openaiErr.Status = status - case float64: - openaiErr.Status = int(status) - case json.Number: - if value, parseErr := status.Int64(); parseErr == nil { - openaiErr.Status = int(value) - } - } - } return openaiErr case string: // 处理简单字符串错误 diff --git a/relay/channel/openai/image_stream_test.go b/relay/channel/openai/image_stream_test.go index c72c15169e7f..b060bbc45e6b 100644 --- a/relay/channel/openai/image_stream_test.go +++ b/relay/channel/openai/image_stream_test.go @@ -153,12 +153,11 @@ func TestOpenaiHandlerWithUsageReturnsImageJSONError(t *testing.T) { usage, err := OpenaiHandlerWithUsage(c, info, resp) require.Nil(t, usage) require.NotNil(t, err) - require.Equal(t, http.StatusBadGateway, err.StatusCode) + require.Equal(t, http.StatusOK, err.StatusCode) oaiError := err.ToOpenAIError() require.Equal(t, "content moderation failed", oaiError.Message) require.Equal(t, "upstream_error", oaiError.Type) require.Equal(t, "content_moderation_failed", oaiError.Code) - require.Equal(t, 502, oaiError.Status) require.Empty(t, recorder.Body.String()) } @@ -186,7 +185,7 @@ func TestOpenaiImageStreamHandlerReturnsJSONErrorFallback(t *testing.T) { usage, err := OpenaiImageStreamHandler(c, info, resp) require.Nil(t, usage) require.NotNil(t, err) - require.Equal(t, http.StatusBadGateway, err.StatusCode) + require.Equal(t, http.StatusOK, err.StatusCode) oaiError := err.ToOpenAIError() require.Equal(t, "image edit failed", oaiError.Message) require.Empty(t, recorder.Body.String()) diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 1613b7d6bb16..8f90eeda17a6 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -570,7 +570,7 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h } if oaiError := usageResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { - return nil, types.WithOpenAIError(*oaiError, openAIErrorStatusCode(oaiError, resp.StatusCode)) + return nil, types.WithOpenAIError(*oaiError, resp.StatusCode) } // 写入新的 response body @@ -581,16 +581,6 @@ func OpenaiHandlerWithUsage(c *gin.Context, info *relaycommon.RelayInfo, resp *h return &usageResp.Usage, nil } -func openAIErrorStatusCode(oaiError *types.OpenAIError, fallback int) int { - if oaiError != nil && oaiError.Status >= http.StatusBadRequest && oaiError.Status <= 599 { - return oaiError.Status - } - if fallback >= http.StatusBadRequest && fallback <= 599 { - return fallback - } - return http.StatusBadGateway -} - func normalizeOpenAIUsage(usage *dto.Usage) { if usage == nil { return @@ -773,7 +763,7 @@ func OpenaiImageJSONAsStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, var usageResp dto.SimpleResponse _ = common.Unmarshal(responseBody, &usageResp) if oaiError := usageResp.GetOpenAIError(); oaiError != nil && oaiError.Type != "" { - return nil, types.WithOpenAIError(*oaiError, openAIErrorStatusCode(oaiError, resp.StatusCode)) + return nil, types.WithOpenAIError(*oaiError, resp.StatusCode) } normalizeOpenAIUsage(&usageResp.Usage) applyUsagePostProcessing(info, &usageResp.Usage, responseBody) diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index 9c24a893f6e0..1d44b80443cd 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -27,8 +27,7 @@ const ( DefaultPingInterval = 10 * time.Second ) -// GetScannerBufferSize returns the configured maximum SSE scanner token size. -func GetScannerBufferSize() int { +func getScannerBufferSize() int { if constant.StreamScannerMaxBufferMB > 0 { return constant.StreamScannerMaxBufferMB << 20 } @@ -41,6 +40,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon return } + // 无条件新建 StreamStatus info.StreamStatus = relaycommon.NewStreamStatus() // 确保响应体总是被关闭 @@ -104,7 +104,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon close(stopChan) }() - scanner.Buffer(make([]byte, InitialScannerBufferSize), GetScannerBufferSize()) + scanner.Buffer(make([]byte, InitialScannerBufferSize), getScannerBufferSize()) scanner.Split(bufio.ScanLines) SetEventStreamHeaders(c) diff --git a/types/error.go b/types/error.go index e4c0a1380ac1..9717401ae7b2 100644 --- a/types/error.go +++ b/types/error.go @@ -11,12 +11,11 @@ import ( ) type OpenAIError struct { - Message string `json:"message"` - Type string `json:"type"` - Param string `json:"param"` - Code any `json:"code"` - Status int `json:"status,omitempty"` - Metadata json.RawMessage `json:"metadata,omitempty"` + Message string `json:"message"` + Type string `json:"type"` + Param string `json:"param"` + Code any `json:"code"` + Metadata json.RawMessage `json:"metadata,omitempty"` } type ClaudeError struct {