mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
Merge pull request #3880 from Wei-Shaw/fix/compact-body-signal-sse-bridge
fix(compact): body-signal 客户端流式请求的响应合成回 SSE（修复 #3875）
This commit is contained in:
Wesley Liddick
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user