Merge pull request #3804 from fengshao1227/fix/compact-body-signal-routing

fix(compact): 识别 /v1/responses body 中的 compaction_trigger 信号

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)
+}

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

backend/internal/service/openai_compact_body_signal.go

@@ -0,0 +1,34 @@
+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"
+//
+// 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
+	}
+	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
+}

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))
+}