mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-11-11 14:32:36 +00:00
shaw
16c0613b21
v0.1.147 的 body-signal 提升（#3804）把 Codex remote compact v2 的流式 /responses 请求改写为上游 unary /responses/compact（JSON），但把 JSON 文档原样写回给了按 Responses SSE 协议消费的客户端。Codex 只从 response.output_item.done 收集 item 且必须收到 response.completed， 收到 JSON 后报 "stream disconnected before completion: stream closed before response.completed" 无限重连，每次重连再白烧一次上游 compact 配额（#3875，v0.1.146 无提升逻辑故正常）。 修复保留提升的全部收益（requireCompact 调度过滤、compact 模型映射、 白名单归一化、上游 unary JSON），补上协议转换缺失的响应半程： - handler 在 body-signal 提升时记录原始 body 的 stream:true 意图 （MarkOpenAICompactClientStream）；path-based 请求不标记，Codex v1 unary 协议与链式 sub2api 的 JSON 写回行为保持不变。 - 四个非流式写回点（handleNonStreamingResponse / handleSSEToJSON / handleNonStreamingResponsePassthrough / handlePassthroughSSEToJSON） 对已标记的 2xx JSON 响应经 writeOpenAICompactSSEBridge 合成最小 Responses SSE：每个 output[] item 一条 response.output_item.done （原始字段逐字节保留），最后一条 response.completed 携带完整 response 对象。 - 按 codex-rs 解析器硬约束兜底：先 json.Compact 防 pretty JSON 换行 破坏 SSE 帧；response.id 缺失时注入 resp_*（必填 string）；usage 缺 input_tokens/output_tokens/total_tokens 整数字段时整体删除， 避免整条 completed 事件解析失败。 - 非 2xx 一律保持 JSON 原样，Codex 依赖 HTTP 状态码走重试链路。 新增 9 个 service 测试 + 4 个 handler 测试覆盖合成形态、id/usage 兜底、 path-based 不受影响、主链路/透传链路/上游 SSE 提取再合成三条端到端。 Fixes #3875 Refs #3777
287 lines
12 KiB
Go
287 lines
12 KiB
Go
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)
|
||
}
|