diff --git a/backend/internal/handler/openai_gateway_compact_body_signal_test.go b/backend/internal/handler/openai_gateway_compact_body_signal_test.go index e7d054d185..a4bfb90466 100644 --- a/backend/internal/handler/openai_gateway_compact_body_signal_test.go +++ b/backend/internal/handler/openai_gateway_compact_body_signal_test.go @@ -117,3 +117,46 @@ func TestNormalizeOpenAIResponsesCompactRequest_SubpathNotPromoted(t *testing.T) require.Equal(t, "/v1/responses/resp_123/cancel", c.Request.URL.Path) require.Equal(t, body, normalized) } + +// 回归 #3875:body-signal 原始请求 stream:true 时必须标记 client-stream, +// 供响应写回阶段把上游 unary JSON 合成回 Codex remote compact v2 所需的 SSE。 +func TestNormalizeOpenAIResponsesCompactRequest_BodySignalStreamTrueMarksClientStream(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"compaction_trigger"}]}`) + c := newCompactBodySignalTestContext(t, "/v1/responses", body) + + _, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + + marked, exists := c.Get(service.OpenAICompactClientStreamKeyForTest()) + require.True(t, exists) + require.Equal(t, true, marked) +} + +func TestNormalizeOpenAIResponsesCompactRequest_BodySignalStreamFalseNotMarked(t *testing.T) { + h := &OpenAIGatewayHandler{} + for name, body := range map[string][]byte{ + "stream_false": []byte(`{"model":"gpt-5.5","stream":false,"input":[{"type":"compaction_trigger"}]}`), + "stream_absent": []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`), + } { + c := newCompactBodySignalTestContext(t, "/v1/responses", body) + _, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok, name) + require.Equal(t, "/v1/responses/compact", c.Request.URL.Path, name) + _, exists := c.Get(service.OpenAICompactClientStreamKeyForTest()) + require.False(t, exists, "case %s 不应标记 client-stream", name) + } +} + +// path-based compact(Codex v1 unary 协议)即使 body 带 stream:true 也不标记, +// 保持 JSON 写回行为不变。 +func TestNormalizeOpenAIResponsesCompactRequest_PathBasedStreamTrueNotMarked(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"message","role":"user","content":"hello"}]}`) + c := newCompactBodySignalTestContext(t, "/v1/responses/compact", body) + + _, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + _, exists := c.Get(service.OpenAICompactClientStreamKeyForTest()) + require.False(t, exists) +} diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index f049711b7f..feba261103 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -585,7 +585,14 @@ func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Con if !isCompactRequest && isBareOpenAIResponsesPath(c) && service.HasCompactionTriggerInInput(body) { c.Request.URL.Path = strings.TrimRight(c.Request.URL.Path, "/") + "/compact" isCompactRequest = true - reqLog.Info("codex.remote_compact.detected_body_signal") + // Codex remote compact v2 的原始请求是流式 /responses:白名单归一化会删除 + // stream 并让上游走 unary JSON,但客户端仍按 SSE 消费响应。记录原始 + // stream 意图,响应写回阶段据此把 JSON 合成回 SSE(#3875)。 + clientStream := gjson.GetBytes(body, "stream").Bool() + if clientStream { + service.MarkOpenAICompactClientStream(c) + } + reqLog.Info("codex.remote_compact.detected_body_signal", zap.Bool("client_stream", clientStream)) } if !isCompactRequest { return body, true diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go new file mode 100644 index 0000000000..ec92e581a4 --- /dev/null +++ b/backend/internal/service/openai_compact_stream_bridge.go @@ -0,0 +1,149 @@ +package service + +import ( + "bytes" + "encoding/json" + "strings" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +// openAICompactClientStreamKey 标记 body-signal compact 请求(Codex remote +// compact v2,见 #3777)的原始 body 携带 stream:true。白名单归一化会删除 +// stream 字段并让上游走 unary /responses/compact(JSON),但客户端仍按 +// Responses SSE 协议消费响应:它必须收到 response.output_item.done(其中恰好 +// 一个 type=compaction 的 item)和 response.completed,否则报 +// "stream closed before response.completed" 并无限重连(#3875)。 +const openAICompactClientStreamKey = "openai_compact_client_stream" + +// MarkOpenAICompactClientStream 由 handler 在 body-signal 提升时调用,记录 +// 客户端的原始 stream 意图,供响应写回阶段决定是否合成 SSE。 +func MarkOpenAICompactClientStream(c *gin.Context) { + if c == nil { + return + } + c.Set(openAICompactClientStreamKey, true) +} + +func OpenAICompactClientStreamKeyForTest() string { + return openAICompactClientStreamKey +} + +func openAICompactClientWantsStream(c *gin.Context) bool { + if c == nil { + return false + } + value, ok := c.Get(openAICompactClientStreamKey) + if !ok { + return false + } + wants, _ := value.(bool) + return wants +} + +// writeOpenAICompactSSEBridge 将 unary compact 的最终 JSON 响应按 Codex remote +// compact v2 的消费协议合成为最小 Responses SSE 流写回客户端。仅当请求被标记 +// 为 body-signal 客户端流式、状态码为 2xx 且 body 是合法 JSON 对象时生效; +// 返回 false 表示未写出任何内容,调用方应按原路径写回。 +func writeOpenAICompactSSEBridge(c *gin.Context, statusCode int, finalResponse []byte) bool { + if c == nil || statusCode < 200 || statusCode >= 300 || !openAICompactClientWantsStream(c) { + return false + } + payload, ok := buildOpenAICompactSSEPayload(finalResponse) + if !ok { + return false + } + header := c.Writer.Header() + header.Set("Content-Type", "text/event-stream") + header.Set("Cache-Control", "no-cache") + header.Set("Connection", "keep-alive") + header.Set("X-Accel-Buffering", "no") + c.Writer.WriteHeader(statusCode) + _, _ = c.Writer.Write(payload) + c.Writer.Flush() + return true +} + +// buildOpenAICompactSSEPayload 把 compact 的 Response JSON 转成 SSE 事件序列: +// 每个 output[] item 一条 response.output_item.done,最后一条 response.completed +// 携带完整 response 对象。Codex 的 SSE 解析只从 output_item.done 收集 item, +// 并要求 response.completed 的 response.id 必填、usage(若存在)必须携带 +// input_tokens/output_tokens/total_tokens 整数字段,否则整条 completed 事件 +// 解析失败,故此处做兜底修补。 +func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) { + if len(finalResponse) == 0 || !gjson.ValidBytes(finalResponse) { + return nil, false + } + if !gjson.ParseBytes(finalResponse).IsObject() { + return nil, false + } + // SSE 的 data 行不允许出现裸换行:上游 JSON 可能是 pretty-printed 形态, + // 嵌入前必须压缩为单行。 + var compacted bytes.Buffer + if err := json.Compact(&compacted, finalResponse); err != nil { + return nil, false + } + response := compacted.Bytes() + root := gjson.ParseBytes(response) + if strings.TrimSpace(root.Get("id").String()) == "" { + next, err := sjson.SetBytes(response, "id", "resp_"+strings.ReplaceAll(uuid.NewString(), "-", "")) + if err != nil { + return nil, false + } + response = next + } + if usage := gjson.GetBytes(response, "usage"); usage.Exists() && !openAICompactUsageParsableByCodex(usage) { + next, err := sjson.DeleteBytes(response, "usage") + if err != nil { + return nil, false + } + response = next + } + + var buf bytes.Buffer + outputIndex := 0 + appendEvent := func(eventType string, data []byte) { + buf.WriteString("event: ") + buf.WriteString(eventType) + buf.WriteString("\ndata: ") + buf.Write(data) + buf.WriteString("\n\n") + } + for _, item := range gjson.GetBytes(response, "output").Array() { + if !item.IsObject() { + continue + } + event, err := sjson.SetBytes([]byte(`{"type":"response.output_item.done"}`), "output_index", outputIndex) + if err != nil { + return nil, false + } + event, err = sjson.SetRawBytes(event, "item", []byte(item.Raw)) + if err != nil { + return nil, false + } + appendEvent("response.output_item.done", event) + outputIndex++ + } + + completed, err := sjson.SetRawBytes([]byte(`{"type":"response.completed"}`), "response", response) + if err != nil { + return nil, false + } + appendEvent("response.completed", completed) + return buf.Bytes(), true +} + +func openAICompactUsageParsableByCodex(usage gjson.Result) bool { + if !usage.IsObject() { + return false + } + for _, field := range []string{"input_tokens", "output_tokens", "total_tokens"} { + if usage.Get(field).Type != gjson.Number { + return false + } + } + return true +} diff --git a/backend/internal/service/openai_compact_stream_bridge_test.go b/backend/internal/service/openai_compact_stream_bridge_test.go new file mode 100644 index 0000000000..2eb5ba80ee --- /dev/null +++ b/backend/internal/service/openai_compact_stream_bridge_test.go @@ -0,0 +1,286 @@ +package service + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func newCompactBridgeTestContext(t *testing.T, markClientStream bool) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", nil) + if markClientStream { + MarkOpenAICompactClientStream(c) + } + return c, rec +} + +func newCompactBridgeTestService() *OpenAIGatewayService { + cfg := &config.Config{} + return &OpenAIGatewayService{ + cfg: cfg, + toolCorrector: NewCodexToolCorrector(), + } +} + +// parseCompactBridgeSSE 把合成的 SSE 文本拆成 (eventType, dataJSON) 序列。 +func parseCompactBridgeSSE(t *testing.T, body string) [][2]string { + t.Helper() + var events [][2]string + for _, block := range strings.Split(strings.TrimSpace(body), "\n\n") { + lines := strings.Split(block, "\n") + require.Len(t, lines, 2, "每个 SSE 事件应为 event+data 两行: %q", block) + require.True(t, strings.HasPrefix(lines[0], "event: "), "缺少 event 行: %q", block) + require.True(t, strings.HasPrefix(lines[1], "data: "), "缺少 data 行: %q", block) + events = append(events, [2]string{ + strings.TrimPrefix(lines[0], "event: "), + strings.TrimPrefix(lines[1], "data: "), + }) + } + return events +} + +func TestBuildOpenAICompactSSEPayload_EmitsItemsAndCompleted(t *testing.T) { + finalResponse := []byte(`{ + "id":"resp_compact_1", + "object":"response", + "model":"gpt-5.1-codex", + "status":"completed", + "output":[ + {"id":"cmp_1","type":"compaction","status":"completed","encrypted_content":"compact-payload","summary":[{"type":"summary_text","text":"compact summary"}],"opaque":{"kept":true}}, + {"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"done"}]} + ], + "usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13} + }`) + + payload, ok := buildOpenAICompactSSEPayload(finalResponse) + require.True(t, ok) + + events := parseCompactBridgeSSE(t, string(payload)) + require.Len(t, events, 3) + + require.Equal(t, "response.output_item.done", events[0][0]) + first := events[0][1] + require.Equal(t, "response.output_item.done", gjson.Get(first, "type").String()) + require.Equal(t, int64(0), gjson.Get(first, "output_index").Int()) + require.Equal(t, "compaction", gjson.Get(first, "item.type").String()) + require.Equal(t, "cmp_1", gjson.Get(first, "item.id").String()) + require.Equal(t, "compact-payload", gjson.Get(first, "item.encrypted_content").String()) + require.Equal(t, "compact summary", gjson.Get(first, "item.summary.0.text").String()) + require.True(t, gjson.Get(first, "item.opaque.kept").Bool(), "item 原始字段必须逐字节保留") + + require.Equal(t, "response.output_item.done", events[1][0]) + require.Equal(t, int64(1), gjson.Get(events[1][1], "output_index").Int()) + require.Equal(t, "message", gjson.Get(events[1][1], "item.type").String()) + + require.Equal(t, "response.completed", events[2][0]) + completed := events[2][1] + require.Equal(t, "response.completed", gjson.Get(completed, "type").String()) + require.Equal(t, "resp_compact_1", gjson.Get(completed, "response.id").String()) + require.Equal(t, int64(13), gjson.Get(completed, "response.usage.total_tokens").Int()) + require.Len(t, gjson.Get(completed, "response.output").Array(), 2) +} + +func TestBuildOpenAICompactSSEPayload_InjectsMissingResponseID(t *testing.T) { + payload, ok := buildOpenAICompactSSEPayload([]byte(`{"output":[{"type":"compaction","encrypted_content":"x"}]}`)) + require.True(t, ok) + + events := parseCompactBridgeSSE(t, string(payload)) + require.Len(t, events, 2) + completed := events[1][1] + // Codex 的 ResponseCompleted 解析要求 response.id 为非空 string,缺失时必须注入。 + id := gjson.Get(completed, "response.id").String() + require.True(t, strings.HasPrefix(id, "resp_"), "缺失 id 必须注入 resp_* 兜底: %q", id) + require.NotEqual(t, "resp_", id) +} + +func TestBuildOpenAICompactSSEPayload_DropsMalformedUsage(t *testing.T) { + payload, ok := buildOpenAICompactSSEPayload([]byte(`{ + "id":"resp_1", + "output":[{"type":"compaction","encrypted_content":"x"}], + "usage":{"prompt_tokens":9,"completion_tokens":4} + }`)) + require.True(t, ok) + + events := parseCompactBridgeSSE(t, string(payload)) + completed := events[len(events)-1][1] + // usage 缺少 Codex 必需的整数字段时必须整体删除,否则 completed 事件解析失败。 + require.False(t, gjson.Get(completed, "response.usage").Exists()) +} + +func TestBuildOpenAICompactSSEPayload_KeepsWellFormedUsage(t *testing.T) { + payload, ok := buildOpenAICompactSSEPayload([]byte(`{ + "id":"resp_1", + "output":[{"type":"compaction","encrypted_content":"x"}], + "usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13,"input_tokens_details":{"cached_tokens":2}} + }`)) + require.True(t, ok) + + events := parseCompactBridgeSSE(t, string(payload)) + completed := events[len(events)-1][1] + require.Equal(t, int64(9), gjson.Get(completed, "response.usage.input_tokens").Int()) + require.Equal(t, int64(2), gjson.Get(completed, "response.usage.input_tokens_details.cached_tokens").Int()) +} + +func TestBuildOpenAICompactSSEPayload_RejectsNonJSONObject(t *testing.T) { + for name, body := range map[string][]byte{ + "empty": nil, + "sse_text": []byte("data: {\"type\":\"response.completed\"}\n\n"), + "array": []byte(`[{"id":"resp_1"}]`), + "non_json": []byte("upstream said no"), + "bare_true": []byte("true"), + } { + _, ok := buildOpenAICompactSSEPayload(body) + require.False(t, ok, "case %s 不应被合成为 SSE", name) + } +} + +func TestWriteOpenAICompactSSEBridge_RequiresMarkAndSuccessStatus(t *testing.T) { + finalResponse := []byte(`{"id":"resp_1","output":[{"type":"compaction","encrypted_content":"x"}]}`) + + // 未标记 client stream:不写出,走原 JSON 路径。 + c, rec := newCompactBridgeTestContext(t, false) + require.False(t, writeOpenAICompactSSEBridge(c, http.StatusOK, finalResponse)) + require.Zero(t, rec.Body.Len()) + + // 标记但上游非 2xx:错误响应保持 JSON 原样(Codex 依赖 HTTP 状态码走重试)。 + c, rec = newCompactBridgeTestContext(t, true) + require.False(t, writeOpenAICompactSSEBridge(c, http.StatusBadGateway, finalResponse)) + require.Zero(t, rec.Body.Len()) + + // 标记且 2xx:合成 SSE。 + c, rec = newCompactBridgeTestContext(t, true) + require.True(t, writeOpenAICompactSSEBridge(c, http.StatusOK, finalResponse)) + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type")) + require.Contains(t, rec.Body.String(), "event: response.completed") +} + +// 回归 #3875:body-signal 提升后的 compact 请求,上游返回 unary JSON, +// 客户端(Codex remote compact v2)必须收到 SSE 事件流而非 JSON 文档, +// 否则报 "stream closed before response.completed" 并无限重连。 +func TestHandleNonStreamingResponse_CompactClientStreamBridgesToSSE(t *testing.T) { + svc := newCompactBridgeTestService() + c, rec := newCompactBridgeTestContext(t, true) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{ + "id":"resp_compact_json", + "object":"response", + "model":"gpt-5.1-codex", + "status":"completed", + "output":[{"id":"cmp_1","type":"compaction","status":"completed","encrypted_content":"compact-payload"}], + "usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13} + }`)), + } + + result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5") + require.NoError(t, err) + require.NotNil(t, result) + + require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type")) + events := parseCompactBridgeSSE(t, rec.Body.String()) + require.Len(t, events, 2) + require.Equal(t, "response.output_item.done", events[0][0]) + require.Equal(t, "compaction", gjson.Get(events[0][1], "item.type").String()) + require.Equal(t, "response.completed", events[1][0]) + require.Equal(t, "resp_compact_json", gjson.Get(events[1][1], "response.id").String()) + + // 计费与响应元数据不受写回形态影响。 + require.NotNil(t, result.usage) + require.Equal(t, 9, result.usage.InputTokens) + require.Equal(t, 4, result.usage.OutputTokens) + require.Equal(t, "resp_compact_json", result.responseID) +} + +// 回归防护:path-based compact(Codex v1 unary 协议、链式 sub2api)未标记 +// client stream,必须保持 v0.1.146 以来的 JSON 写回行为。 +func TestHandleNonStreamingResponse_PathBasedCompactStaysJSON(t *testing.T) { + svc := newCompactBridgeTestService() + c, rec := newCompactBridgeTestContext(t, false) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{ + "id":"resp_compact_json", + "output":[{"id":"cmp_1","type":"compaction","encrypted_content":"compact-payload"}], + "usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13} + }`)), + } + + result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5") + require.NoError(t, err) + require.NotNil(t, result) + + require.NotContains(t, rec.Header().Get("Content-Type"), "text/event-stream") + body := rec.Body.String() + require.Equal(t, "resp_compact_json", gjson.Get(body, "id").String()) + require.Equal(t, "compaction", gjson.Get(body, "output.0.type").String()) +} + +// 上游对 compact 返回 SSE(如链式网关)时,最终响应经 SSE→JSON 提取后, +// 对 client-stream 请求同样必须再合成回 SSE。 +func TestHandleSSEToJSON_CompactClientStreamBridgesToSSE(t *testing.T) { + svc := newCompactBridgeTestService() + c, rec := newCompactBridgeTestContext(t, true) + upstreamSSE := strings.Join([]string{ + `data: {"type":"response.completed","response":{"id":"resp_compact_sse","object":"response","model":"gpt-5.1-codex","status":"completed","output":[{"id":"cmp_sse_1","type":"compaction","status":"completed","encrypted_content":"compact-sse-payload"}],"usage":{"input_tokens":3,"output_tokens":2,"total_tokens":5}}}`, + "", + }, "\n") + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamSSE)), + } + + result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5") + require.NoError(t, err) + require.NotNil(t, result) + + require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type")) + events := parseCompactBridgeSSE(t, rec.Body.String()) + require.Len(t, events, 2) + require.Equal(t, "response.output_item.done", events[0][0]) + require.Equal(t, "compact-sse-payload", gjson.Get(events[0][1], "item.encrypted_content").String()) + require.Equal(t, "response.completed", events[1][0]) + require.Equal(t, "resp_compact_sse", gjson.Get(events[1][1], "response.id").String()) +} + +// 透传分支(OAuth passthrough)同样命中桥接。 +func TestHandleNonStreamingResponsePassthrough_CompactClientStreamBridgesToSSE(t *testing.T) { + svc := newCompactBridgeTestService() + c, rec := newCompactBridgeTestContext(t, true) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{ + "id":"resp_compact_pt", + "output":[{"id":"cmp_pt_1","type":"compaction","encrypted_content":"compact-pt-payload"}], + "usage":{"input_tokens":7,"output_tokens":3,"total_tokens":10} + }`)), + } + + result, err := svc.handleNonStreamingResponsePassthrough(context.Background(), resp, c, "gpt-5.5", "") + require.NoError(t, err) + require.NotNil(t, result) + + require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type")) + events := parseCompactBridgeSSE(t, rec.Body.String()) + require.Len(t, events, 2) + require.Equal(t, "compaction", gjson.Get(events[0][1], "item.type").String()) + require.Equal(t, "resp_compact_pt", gjson.Get(events[1][1], "response.id").String()) + require.NotNil(t, result.usage) + require.Equal(t, 7, result.usage.InputTokens) +} diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 635b2ac950..3a56ab5c61 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -1088,7 +1088,9 @@ func (s *OpenAIGatewayService) handleNonStreamingResponsePassthrough( if originalModel != "" && mappedModel != "" && originalModel != mappedModel { body = s.replaceModelInResponseBody(body, mappedModel, originalModel) } - c.Data(resp.StatusCode, contentType, body) + if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { + c.Data(resp.StatusCode, contentType, body) + } return &openaiNonStreamingResultPassthrough{ OpenAIUsage: usage, usage: usage, @@ -1151,7 +1153,9 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c contentType = "text/event-stream" } } - c.Data(resp.StatusCode, contentType, body) + if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { + c.Data(resp.StatusCode, contentType, body) + } return &openaiNonStreamingResultPassthrough{ OpenAIUsage: usage, diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index b814ff96f3..cb4b780cdf 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -825,7 +825,9 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r } } - c.Data(resp.StatusCode, contentType, body) + if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { + c.Data(resp.StatusCode, contentType, body) + } return &openaiNonStreamingResult{ OpenAIUsage: usage, @@ -907,7 +909,9 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte contentType = "text/event-stream" } } - c.Data(resp.StatusCode, contentType, body) + if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { + c.Data(resp.StatusCode, contentType, body) + } return &openaiNonStreamingResult{ OpenAIUsage: usage,