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 0000000000..e7d054d185 --- /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 551de2dd61..45eafbae76 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 fa0e44e5bf..fce62046c1 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 78543cb08e..8e05f4f98b 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)) }