From 2dd2be9922843ed08fbbd3eac310dbe0d3271f8b Mon Sep 17 00:00:00 2001 From: li Date: Tue, 7 Jul 2026 21:23:02 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(compact):=20=E8=AF=86=E5=88=AB=20/v1/re?= =?UTF-8?q?sponses=20body=20=E4=B8=AD=E7=9A=84=20compaction=5Ftrigger=20?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex remote compact v2 可以把 compact 触发器作为 input item （type=compaction_trigger）嵌入普通 POST /v1/responses 请求体， 而非直接调用 /v1/responses/compact。此前 isCompactRequest 仅检查 URL path 后缀，导致 body-signal 形式的 compact 请求被当作普通 Responses 处理——上游路径、模型映射、body 归一化全部错误，Codex 收到非 compact 响应后报 "expected exactly one compaction output item, got 0"，长会话无法继续。 新增 hasCompactionTriggerInInput 检测 input 中的 compaction_trigger item，命中后提升为 compact 请求并改写 URL path，使后续所有 isCompactRequest 分支（模型映射、body 归一化、上游 URL 构建） 自动生效。 Fixes #3777 (part 1: body-signal routing) --- .../service/openai_compact_body_signal.go | 30 ++++++++++ .../openai_compact_body_signal_test.go | 56 +++++++++++++++++++ .../service/openai_gateway_service.go | 13 +++++ 3 files changed, 99 insertions(+) create mode 100644 backend/internal/service/openai_compact_body_signal.go create mode 100644 backend/internal/service/openai_compact_body_signal_test.go diff --git a/backend/internal/service/openai_compact_body_signal.go b/backend/internal/service/openai_compact_body_signal.go new file mode 100644 index 000000000..fa0e44e5b --- /dev/null +++ b/backend/internal/service/openai_compact_body_signal.go @@ -0,0 +1,30 @@ +package service + +import "github.com/tidwall/gjson" + +// hasCompactionTriggerInInput detects the Codex remote compact v2 body signal: +// an input item with type "compaction_trigger". When the client sends this +// inside a normal POST /v1/responses (instead of POST /v1/responses/compact), +// the request must still be treated as a compact request — otherwise the +// upstream path, model mapping, and body normalization are all wrong, causing +// Codex to receive a non-compact response and fail with: +// +// "remote compaction v2 expected exactly one compaction output item, got 0" +func hasCompactionTriggerInInput(body []byte) bool { + if len(body) == 0 { + return false + } + input := gjson.GetBytes(body, "input") + if !input.IsArray() { + return false + } + found := false + input.ForEach(func(_, item gjson.Result) bool { + if item.Get("type").String() == "compaction_trigger" { + found = true + return false + } + return true + }) + return found +} diff --git a/backend/internal/service/openai_compact_body_signal_test.go b/backend/internal/service/openai_compact_body_signal_test.go new file mode 100644 index 000000000..78543cb08 --- /dev/null +++ b/backend/internal/service/openai_compact_body_signal_test.go @@ -0,0 +1,56 @@ +//go:build unit + +package service + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestHasCompactionTriggerInInput_DetectsCompactSignal(t *testing.T) { + body := []byte(`{ + "model":"gpt-5.5", + "stream":true, + "input":[ + {"type":"message","role":"user","content":"hello"}, + {"type":"compaction_trigger"} + ] + }`) + require.True(t, hasCompactionTriggerInInput(body)) +} + +func TestHasCompactionTriggerInInput_NoTrigger(t *testing.T) { + body := []byte(`{ + "model":"gpt-5.5", + "input":[ + {"type":"message","role":"user","content":"hello"} + ] + }`) + require.False(t, hasCompactionTriggerInInput(body)) +} + +func TestHasCompactionTriggerInInput_EmptyInput(t *testing.T) { + body := []byte(`{"model":"gpt-5.5","input":[]}`) + require.False(t, hasCompactionTriggerInInput(body)) +} + +func TestHasCompactionTriggerInInput_NoInputField(t *testing.T) { + body := []byte(`{"model":"gpt-5.5"}`) + require.False(t, hasCompactionTriggerInInput(body)) +} + +func TestHasCompactionTriggerInInput_EmptyBody(t *testing.T) { + require.False(t, hasCompactionTriggerInInput(nil)) + require.False(t, hasCompactionTriggerInInput([]byte{})) +} + +func TestHasCompactionTriggerInInput_StringInput(t *testing.T) { + body := []byte(`{"model":"gpt-5.5","input":"compaction_trigger"}`) + require.False(t, hasCompactionTriggerInInput(body)) +} + +func TestHasCompactionTriggerInInput_CompactTriggerOnly(t *testing.T) { + body := []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`) + require.True(t, hasCompactionTriggerInInput(body)) +} diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index de3192ae5..e4e1e14fe 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -2777,6 +2777,19 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } upstreamModel := billingModel isCompactRequest := isOpenAIResponsesCompactPath(c) + // Body-signal detection: Codex remote compact v2 can send the compact + // trigger inside a normal POST /v1/responses body (input item with + // type "compaction_trigger") instead of calling /v1/responses/compact + // directly. Detect and promote to compact request (#3777). + if !isCompactRequest && hasCompactionTriggerInInput(body) { + isCompactRequest = true + // Rewrite the request path so downstream URL builders + // (appendOpenAIResponsesRequestPathSuffix) route to /responses/compact. + if c != nil && c.Request != nil && c.Request.URL != nil { + c.Request.URL.Path = strings.TrimRight(c.Request.URL.Path, "/") + "/compact" + } + logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Compact body-signal detected: input contains compaction_trigger (account: %s)", account.Name) + } compactMapped := false if isCompactRequest { compactMappedModel := resolveOpenAICompactForwardModel(account, billingModel) From a56eb5b4dc6aafeea364e8296a8be1a53e4f04dc Mon Sep 17 00:00:00 2001 From: shaw Date: Wed, 8 Jul 2026 10:04:14 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(compact):=20body-signal=20=E6=8F=90?= =?UTF-8?q?=E5=8D=87=E4=B8=8A=E7=A7=BB=E5=88=B0=20handler=20=E5=B1=82?= =?UTF-8?q?=E5=B9=B6=E5=AF=B9=E9=BD=90=20path-based=20=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 合并 main 解决拆分冲突后，将原先 Forward 内的 body-signal 提升重构到 handler 的 compact 归一化入口之前，修复原方案的四个问题： - reqStream 未重推导：body-signal 原始请求带 stream:true，Forward 级提升 后 compact 上游返回 JSON（Accept: application/json）却被流式 handler 解析，"stream ended before a terminal event" 会触发最多 max_account_switches 次换号 failover，且每次都白烧一次上游 compact 配额； handler 级提升让白名单归一化先删除 stream，reqStream 自然为 false。 - requireCompact 调度过滤失效：原方案 path 改写发生在 requireCompact 判定 之后，调度器不会过滤不支持 compact 的账号；现在改写先于该判定。 - passthrough / Grok / chat-completions 桥接分支位于 Forward 检测点之前， passthrough 账号完全无法命中；handler 级改写对所有分支生效。 - body 归一化口径不一致：body-signal 现在与 path-based 一样走白名单归一化 （prompt_cache_key 等一并删除），而非仅依赖 OAuth 黑名单转换。 检测函数导出为 HasCompactionTriggerInInput 供 handler 使用，保留原 PR 的 7 个单测；新增 6 个 handler 级回归测试（提升、codex 别名路由、尾斜杠、 子路径不误伤、path-based 无双重后缀、普通请求不受影响）。 Refs #3777 --- ...openai_gateway_compact_body_signal_test.go | 119 ++++++++++++++++++ .../handler/openai_gateway_handler.go | 56 +++++++-- .../service/openai_compact_body_signal.go | 8 +- .../openai_compact_body_signal_test.go | 16 +-- 4 files changed, 177 insertions(+), 22 deletions(-) create mode 100644 backend/internal/handler/openai_gateway_compact_body_signal_test.go diff --git a/backend/internal/handler/openai_gateway_compact_body_signal_test.go b/backend/internal/handler/openai_gateway_compact_body_signal_test.go new file mode 100644 index 000000000..e7d054d18 --- /dev/null +++ b/backend/internal/handler/openai_gateway_compact_body_signal_test.go @@ -0,0 +1,119 @@ +package handler + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" + "go.uber.org/zap" +) + +func newCompactBodySignalTestContext(t *testing.T, path string, body []byte) *gin.Context { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, path, bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + return c +} + +// body-signal 提升后必须与 path-based compact 走同一条链路： +// path 改写、requireCompact 判定、stream/store/prompt_cache_key 归一化删除。 +// 回归防护：若 stream 字段存活，Forward 会用流式 handler 解析 compact 的 +// JSON 响应，导致 "stream ended before a terminal event" 的换号 failover 风暴。 +func TestNormalizeOpenAIResponsesCompactRequest_BodySignalPromoted(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{ + "model":"gpt-5.5", + "stream":true, + "store":true, + "prompt_cache_key":"pck-signal-1", + "input":[ + {"type":"message","role":"user","content":"hello"}, + {"type":"compaction_trigger"} + ] + }`) + c := newCompactBodySignalTestContext(t, "/v1/responses", body) + + normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + + require.Equal(t, "/v1/responses/compact", c.Request.URL.Path) + require.True(t, isOpenAIRemoteCompactPath(c)) + + require.False(t, gjson.GetBytes(normalized, "stream").Exists()) + require.False(t, gjson.GetBytes(normalized, "store").Exists()) + require.False(t, gjson.GetBytes(normalized, "prompt_cache_key").Exists()) + require.Equal(t, "gpt-5.5", gjson.GetBytes(normalized, "model").String()) + require.True(t, gjson.GetBytes(normalized, "input").IsArray()) + + reqStream, streamOK := parseOpenAICompatibleStream(normalized) + require.True(t, streamOK) + require.False(t, reqStream) + + seed, exists := c.Get(service.OpenAICompactSessionSeedKeyForTest()) + require.True(t, exists) + require.Equal(t, "pck-signal-1", seed) +} + +func TestNormalizeOpenAIResponsesCompactRequest_BodySignalTrailingSlash(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []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) + require.Equal(t, "/v1/responses/compact", c.Request.URL.Path) +} + +func TestNormalizeOpenAIResponsesCompactRequest_CodexDirectAliasPromoted(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`) + c := newCompactBodySignalTestContext(t, "/backend-api/codex/responses", body) + + _, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + require.Equal(t, "/backend-api/codex/responses/compact", c.Request.URL.Path) +} + +func TestNormalizeOpenAIResponsesCompactRequest_NoTriggerUntouched(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", body) + + normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + require.Equal(t, "/v1/responses", c.Request.URL.Path) + require.False(t, isOpenAIRemoteCompactPath(c)) + require.Equal(t, body, normalized) + require.True(t, gjson.GetBytes(normalized, "stream").Bool()) +} + +func TestNormalizeOpenAIResponsesCompactRequest_PathBasedNoDoubleSuffix(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{"model":"gpt-5.5","stream":true,"store":true,"input":[{"type":"message","role":"user","content":"hello"}]}`) + c := newCompactBodySignalTestContext(t, "/v1/responses/compact", body) + + normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + require.Equal(t, "/v1/responses/compact", c.Request.URL.Path) + require.False(t, gjson.GetBytes(normalized, "stream").Exists()) + require.False(t, gjson.GetBytes(normalized, "store").Exists()) +} + +func TestNormalizeOpenAIResponsesCompactRequest_SubpathNotPromoted(t *testing.T) { + h := &OpenAIGatewayHandler{} + body := []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`) + c := newCompactBodySignalTestContext(t, "/v1/responses/resp_123/cancel", body) + + normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body) + require.True(t, ok) + require.Equal(t, "/v1/responses/resp_123/cancel", c.Request.URL.Path) + require.Equal(t, body, normalized) +} diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 551de2dd6..45eafbae7 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -202,18 +202,9 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { setOpsRequestContext(c, "", false) sessionHashBody := body - if service.IsOpenAIResponsesCompactPathForTest(c) { - if compactSeed := strings.TrimSpace(gjson.GetBytes(body, "prompt_cache_key").String()); compactSeed != "" { - c.Set(service.OpenAICompactSessionSeedKeyForTest(), compactSeed) - } - normalizedCompactBody, normalizedCompact, compactErr := service.NormalizeOpenAICompactRequestBodyForTest(body) - if compactErr != nil { - h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Failed to normalize compact request body") - return - } - if normalizedCompact { - body = normalizedCompactBody - } + body, ok = h.normalizeOpenAIResponsesCompactRequest(c, reqLog, body) + if !ok { + return } // 校验请求体 JSON 合法性 @@ -573,6 +564,47 @@ func isOpenAIRemoteCompactPath(c *gin.Context) bool { return strings.HasSuffix(normalizedPath, "/responses/compact") } +// isBareOpenAIResponsesPath 仅匹配裸 /responses 端点（无 /compact 等子路径）， +// body-signal 提升只允许发生在这里，避免误伤 /responses/{id}/... 形态的请求。 +func isBareOpenAIResponsesPath(c *gin.Context) bool { + if c == nil || c.Request == nil || c.Request.URL == nil { + return false + } + normalizedPath := strings.TrimRight(strings.TrimSpace(c.Request.URL.Path), "/") + return strings.HasSuffix(normalizedPath, "/responses") +} + +// normalizeOpenAIResponsesCompactRequest 统一处理两种入站 compact 形态： +// path-based（POST /v1/responses/compact）与 Codex remote compact v2 的 +// body-signal（普通 POST /v1/responses 的 input 中携带 type=compaction_trigger， +// 见 #3777）。body-signal 命中时在 stream 解析、compact body 归一化与 +// requireCompact 调度判定之前改写 URL path，使后续全部链路（含 passthrough +// 分支与上游 URL 构建）与 path-based 完全一致。 +// 返回归一化后的 body；ok=false 表示错误响应已写出，调用方应直接 return。 +func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Context, reqLog *zap.Logger, body []byte) ([]byte, bool) { + isCompactRequest := service.IsOpenAIResponsesCompactPathForTest(c) + 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") + } + if !isCompactRequest { + return body, true + } + if compactSeed := strings.TrimSpace(gjson.GetBytes(body, "prompt_cache_key").String()); compactSeed != "" { + c.Set(service.OpenAICompactSessionSeedKeyForTest(), compactSeed) + } + normalizedCompactBody, normalizedCompact, compactErr := service.NormalizeOpenAICompactRequestBodyForTest(body) + if compactErr != nil { + h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Failed to normalize compact request body") + return nil, false + } + if normalizedCompact { + body = normalizedCompactBody + } + return body, true +} + func (h *OpenAIGatewayHandler) logOpenAIRemoteCompactOutcome(c *gin.Context, startedAt time.Time) { if !isOpenAIRemoteCompactPath(c) { return diff --git a/backend/internal/service/openai_compact_body_signal.go b/backend/internal/service/openai_compact_body_signal.go index fa0e44e5b..fce62046c 100644 --- a/backend/internal/service/openai_compact_body_signal.go +++ b/backend/internal/service/openai_compact_body_signal.go @@ -2,7 +2,7 @@ package service import "github.com/tidwall/gjson" -// hasCompactionTriggerInInput detects the Codex remote compact v2 body signal: +// HasCompactionTriggerInInput detects the Codex remote compact v2 body signal: // an input item with type "compaction_trigger". When the client sends this // inside a normal POST /v1/responses (instead of POST /v1/responses/compact), // the request must still be treated as a compact request — otherwise the @@ -10,7 +10,11 @@ import "github.com/tidwall/gjson" // Codex to receive a non-compact response and fail with: // // "remote compaction v2 expected exactly one compaction output item, got 0" -func hasCompactionTriggerInInput(body []byte) bool { +// +// The gateway handler promotes such requests by rewriting the URL path to the +// compact form before stream parsing, compact body normalization, and +// compact-capable account scheduling, so both inbound forms share one code path. +func HasCompactionTriggerInInput(body []byte) bool { if len(body) == 0 { return false } diff --git a/backend/internal/service/openai_compact_body_signal_test.go b/backend/internal/service/openai_compact_body_signal_test.go index 78543cb08..8e05f4f98 100644 --- a/backend/internal/service/openai_compact_body_signal_test.go +++ b/backend/internal/service/openai_compact_body_signal_test.go @@ -17,7 +17,7 @@ func TestHasCompactionTriggerInInput_DetectsCompactSignal(t *testing.T) { {"type":"compaction_trigger"} ] }`) - require.True(t, hasCompactionTriggerInInput(body)) + require.True(t, HasCompactionTriggerInInput(body)) } func TestHasCompactionTriggerInInput_NoTrigger(t *testing.T) { @@ -27,30 +27,30 @@ func TestHasCompactionTriggerInInput_NoTrigger(t *testing.T) { {"type":"message","role":"user","content":"hello"} ] }`) - require.False(t, hasCompactionTriggerInInput(body)) + require.False(t, HasCompactionTriggerInInput(body)) } func TestHasCompactionTriggerInInput_EmptyInput(t *testing.T) { body := []byte(`{"model":"gpt-5.5","input":[]}`) - require.False(t, hasCompactionTriggerInInput(body)) + require.False(t, HasCompactionTriggerInInput(body)) } func TestHasCompactionTriggerInInput_NoInputField(t *testing.T) { body := []byte(`{"model":"gpt-5.5"}`) - require.False(t, hasCompactionTriggerInInput(body)) + require.False(t, HasCompactionTriggerInInput(body)) } func TestHasCompactionTriggerInInput_EmptyBody(t *testing.T) { - require.False(t, hasCompactionTriggerInInput(nil)) - require.False(t, hasCompactionTriggerInInput([]byte{})) + require.False(t, HasCompactionTriggerInInput(nil)) + require.False(t, HasCompactionTriggerInInput([]byte{})) } func TestHasCompactionTriggerInInput_StringInput(t *testing.T) { body := []byte(`{"model":"gpt-5.5","input":"compaction_trigger"}`) - require.False(t, hasCompactionTriggerInInput(body)) + require.False(t, HasCompactionTriggerInInput(body)) } func TestHasCompactionTriggerInInput_CompactTriggerOnly(t *testing.T) { body := []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`) - require.True(t, hasCompactionTriggerInInput(body)) + require.True(t, HasCompactionTriggerInInput(body)) }