Merge pull request #5641 from InCerryGit/fix/issue-5624-remote-compaction-v2

fix(openai): preserve remote compaction v2 responses endpoint
This commit is contained in:
Wesley Liddick
2026-08-15 13:46:31 +08:00
committed by GitHub
17 changed files with 926 additions and 133 deletions
@@ -25,54 +25,211 @@ func newCompactBodySignalTestContext(t *testing.T, path string, body []byte) *gi
func TestNormalizeOpenAIResponsesCompactRequest_RemoteV2StaysOnResponses(t *testing.T) {
h := &OpenAIGatewayHandler{}
body := []byte(`{
"model":"gpt-5.6-sol",
"stream":true,
"store":true,
"prompt_cache_key":"pck-signal-1",
"reasoning":{"effort":"max","context":"all_turns"},
"input":[
{"type":"message","role":"user","content":"hello"},
{"type":"compaction_trigger"}
]
}`)
c := newCompactBodySignalTestContext(t, "/v1/responses", body)
c.Request.Header.Set("x-codex-beta-features", "responses_websockets_v2, remote_compaction_v2, another_feature")
tests := []struct {
name string
betaHeader string
userAgent string
}{
{name: "headerless"},
{name: "unrelated_header", betaHeader: "responses_websockets_v2"},
{name: "wrong_case_header", betaHeader: "REMOTE_COMPACTION_V2"},
{name: "declared_header", betaHeader: "remote_compaction_v2"},
{name: "codex_cli_user_agent", userAgent: "codex_cli_rs/0.144.1 (Ubuntu 22.4.0; x86_64) xterm-256color"},
{name: "codex_desktop_user_agent", userAgent: "Codex Desktop/0.139.0 (Mac OS X 14; arm64) unknown"},
}
normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
require.True(t, ok)
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
body := []byte(`{
"model":"gpt-5.6-sol",
"stream":true,
"store":true,
"prompt_cache_key":"pck-signal-1",
"reasoning":{"effort":"max","context":"all_turns"},
"input":[
{"type":"message","role":"user","content":"hello"},
{"type":"compaction_trigger"}
]
}`)
c := newCompactBodySignalTestContext(t, "/v1/responses", body)
if tt.betaHeader != "" {
c.Request.Header.Set("x-codex-beta-features", tt.betaHeader)
}
if tt.userAgent != "" {
c.Request.Header.Set("User-Agent", tt.userAgent)
}
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())
require.True(t, gjson.GetBytes(normalized, "store").Bool())
require.Equal(t, "pck-signal-1", gjson.GetBytes(normalized, "prompt_cache_key").String())
require.Equal(t, "max", gjson.GetBytes(normalized, "reasoning.effort").String())
require.Equal(t, "all_turns", gjson.GetBytes(normalized, "reasoning.context").String())
normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
require.True(t, ok)
reqStream, streamOK := parseOpenAICompatibleStream(normalized)
require.True(t, streamOK)
require.True(t, reqStream)
require.Equal(t, "/v1/responses", c.Request.URL.Path)
require.False(t, isOpenAILegacyCompactPath(c))
require.Equal(t, body, normalized)
require.True(t, gjson.GetBytes(normalized, "stream").Bool())
require.True(t, gjson.GetBytes(normalized, "store").Bool())
require.Equal(t, "pck-signal-1", gjson.GetBytes(normalized, "prompt_cache_key").String())
require.Equal(t, "max", gjson.GetBytes(normalized, "reasoning.effort").String())
require.Equal(t, "all_turns", gjson.GetBytes(normalized, "reasoning.context").String())
legacyCompact := service.IsOpenAIResponsesCompactPath(c)
nativeV2 := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(normalized)
require.False(t, legacyCompact)
require.True(t, nativeV2)
require.Equal(t, service.OpenAIEndpointCapabilityResponses,
openAIResponsesRequiredCapabilityForRequest(false, nativeV2 || legacyCompact, service.PlatformOpenAI))
_, seedExists := c.Get(service.OpenAICompactSessionSeedKeyForTest())
require.False(t, seedExists)
_, streamMarkerExists := c.Get(service.OpenAICompactClientStreamKeyForTest())
require.False(t, streamMarkerExists)
reqStream, streamOK := parseOpenAICompatibleStream(normalized)
require.True(t, streamOK)
require.True(t, reqStream)
_, seedExists := c.Get(service.OpenAICompactSessionSeedKeyForTest())
require.False(t, seedExists)
_, streamMarkerExists := c.Get(service.OpenAICompactClientStreamKeyForTest())
require.False(t, streamMarkerExists)
})
}
}
func TestNormalizeOpenAIResponsesCompactRequest_RemoteV2PathAliasesStayOnResponses(t *testing.T) {
h := &OpenAIGatewayHandler{}
body := []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"compaction_trigger"}]}`)
for _, path := range []string{"/v1/responses/", "/backend-api/codex/responses"} {
for _, path := range []string{
"/v1/responses/",
"/openai/v1/responses",
"/responses",
"/backend-api/codex/responses",
} {
t.Run(path, func(t *testing.T) {
c := newCompactBodySignalTestContext(t, path, body)
c.Request.Header.Set("x-codex-beta-features", "remote_compaction_v2")
normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
require.True(t, ok)
require.Equal(t, path, c.Request.URL.Path)
require.Equal(t, body, normalized)
legacyCompact := service.IsOpenAIResponsesCompactPath(c)
nativeV2 := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(normalized)
require.False(t, legacyCompact)
require.True(t, nativeV2)
require.Equal(t, service.OpenAIEndpointCapabilityResponses,
openAIResponsesRequiredCapabilityForRequest(false, nativeV2 || legacyCompact, service.PlatformOpenAI))
})
}
}
func TestOpenAIResponsesCompactionRoutingFlags(t *testing.T) {
h := &OpenAIGatewayHandler{}
tests := []struct {
name string
body []byte
path string
wantLegacyBefore bool
wantNativeBefore bool
wantLegacyAfter bool
wantNativeAfter bool
wantCapabilityAfter service.OpenAIEndpointCapability
wantPathAfter string
wantBodyUnchanged bool
}{
{
name: "native_v2_stream_trigger",
body: []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses",
wantLegacyBefore: false,
wantNativeBefore: true,
wantLegacyAfter: false,
wantNativeAfter: true,
wantCapabilityAfter: service.OpenAIEndpointCapabilityResponses,
wantPathAfter: "/v1/responses",
wantBodyUnchanged: true,
},
{
name: "native_v2_without_trigger",
body: []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"message","role":"user","content":"hello"}]}`),
path: "/v1/responses",
wantLegacyBefore: false,
wantNativeBefore: false,
wantLegacyAfter: false,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityChatCompletions,
wantPathAfter: "/v1/responses",
wantBodyUnchanged: true,
},
{
name: "explicit_compact",
body: []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses/compact",
wantLegacyBefore: true,
wantNativeBefore: false,
wantLegacyAfter: true,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityResponses,
wantPathAfter: "/v1/responses/compact",
},
{
name: "nested_compact",
body: []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses/compact/detail",
wantLegacyBefore: true,
wantNativeBefore: false,
wantLegacyAfter: true,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityResponses,
wantPathAfter: "/v1/responses/compact/detail",
},
{
name: "responses_subpath_with_native_signal",
body: []byte(`{"model":"gpt-5.6-sol","stream":true,"input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses/resp_123/responses",
wantLegacyBefore: false,
wantNativeBefore: false,
wantLegacyAfter: false,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityChatCompletions,
wantPathAfter: "/v1/responses/resp_123/responses",
wantBodyUnchanged: true,
},
{
name: "stream_false_promotes",
body: []byte(`{"model":"gpt-5.6-sol","stream":false,"input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses",
wantLegacyBefore: false,
wantNativeBefore: false,
wantLegacyAfter: true,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityResponses,
wantPathAfter: "/v1/responses/compact",
},
{
name: "stream_absent_promotes",
body: []byte(`{"model":"gpt-5.6-sol","input":[{"type":"compaction_trigger"}]}`),
path: "/v1/responses",
wantLegacyBefore: false,
wantNativeBefore: false,
wantLegacyAfter: true,
wantNativeAfter: false,
wantCapabilityAfter: service.OpenAIEndpointCapabilityResponses,
wantPathAfter: "/v1/responses/compact",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c := newCompactBodySignalTestContext(t, tt.path, tt.body)
legacyBefore := service.IsOpenAIResponsesCompactPath(c)
nativeBefore := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(tt.body)
require.Equal(t, tt.wantLegacyBefore, legacyBefore)
require.Equal(t, tt.wantNativeBefore, nativeBefore)
normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), tt.body)
require.True(t, ok)
require.Equal(t, tt.wantPathAfter, c.Request.URL.Path)
legacyAfter := service.IsOpenAIResponsesCompactPath(c)
nativeAfter := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(normalized)
require.Equal(t, tt.wantLegacyAfter, legacyAfter)
require.Equal(t, tt.wantNativeAfter, nativeAfter)
require.Equal(t, tt.wantCapabilityAfter,
openAIResponsesRequiredCapabilityForRequest(false, nativeAfter || legacyAfter, service.PlatformOpenAI))
if tt.wantBodyUnchanged {
require.Equal(t, tt.body, normalized)
}
})
}
}
@@ -106,31 +263,22 @@ func TestNormalizeOpenAIResponsesCompactRequest_NonRemoteV2BodySignalPromoted(t
wantMarked bool
}{
{
name: "no_header",
body: []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"compaction_trigger"}]}`),
wantMarked: true,
name: "stream_false_headerless",
body: []byte(`{"model":"gpt-5.5","stream":false,"input":[{"type":"compaction_trigger"}]}`),
},
{
name: "unrelated_header",
body: []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"compaction_trigger"}]}`),
betaHeader: "responses_websockets_v2",
wantMarked: true,
name: "stream_absent_headerless",
body: []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`),
},
{
name: "wrong_case_header",
body: []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"compaction_trigger"}]}`),
betaHeader: "REMOTE_COMPACTION_V2",
wantMarked: true,
},
{
name: "stream_false",
name: "stream_false_declared_header",
body: []byte(`{"model":"gpt-5.5","stream":false,"input":[{"type":"compaction_trigger"}]}`),
betaHeader: "remote_compaction_v2",
},
{
name: "stream_absent",
name: "stream_absent_wrong_case_header",
body: []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`),
betaHeader: "remote_compaction_v2",
betaHeader: "REMOTE_COMPACTION_V2",
},
}
@@ -163,7 +311,7 @@ func TestNormalizeOpenAIResponsesCompactRequest_NoTriggerUntouched(t *testing.T)
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.False(t, isOpenAILegacyCompactPath(c))
require.Equal(t, body, normalized)
require.True(t, gjson.GetBytes(normalized, "stream").Bool())
}
@@ -172,7 +320,6 @@ func TestNormalizeOpenAIResponsesCompactRequest_PathBasedNoDoubleSuffix(t *testi
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)
c.Request.Header.Set("x-codex-beta-features", "remote_compaction_v2")
normalized, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
require.True(t, ok)
@@ -105,21 +105,29 @@ func captureHandlerStructuredLog(t *testing.T) (*handlerInMemoryLogSink, func())
}
}
func TestIsOpenAIRemoteCompactPath(t *testing.T) {
require.False(t, isOpenAIRemoteCompactPath(nil))
func TestIsOpenAILegacyCompactPath(t *testing.T) {
require.False(t, isOpenAILegacyCompactPath(nil))
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", nil)
require.True(t, isOpenAIRemoteCompactPath(c))
c.Request = httptest.NewRequest(http.MethodPost, "/responses/compact/", nil)
require.True(t, isOpenAIRemoteCompactPath(c))
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
require.False(t, isOpenAIRemoteCompactPath(c))
for _, test := range []struct {
path string
want bool
}{
{path: "/v1/responses/compact", want: true},
{path: "/v1/responses/compact/detail", want: true},
{path: "/responses/compact/", want: true},
{path: "/v1/responses", want: false},
{path: "/openai/v1/responses", want: false},
{path: "/responses", want: false},
{path: "/backend-api/codex/responses", want: false},
{path: "/v1/responses/resp_123/cancel", want: false},
} {
c.Request = httptest.NewRequest(http.MethodPost, test.path, nil)
require.Equal(t, test.want, isOpenAILegacyCompactPath(c), test.path)
}
}
func TestLogOpenAIRemoteCompactOutcome_Succeeded(t *testing.T) {
@@ -182,6 +182,16 @@ func openAIResponsesRequiredCapability(imageIntent bool, platform string) servic
return service.OpenAIEndpointCapabilityChatCompletions
}
// openAIResponsesRequiredCapabilityForRequest returns the endpoint capability
// required by an image or Responses request. needsResponses includes both the
// legacy /responses/compact endpoint and native remote compaction v2.
func openAIResponsesRequiredCapabilityForRequest(imageIntent bool, needsResponses bool, platform string) service.OpenAIEndpointCapability {
if needsResponses && platform == service.PlatformOpenAI {
return service.OpenAIEndpointCapabilityResponses
}
return openAIResponsesRequiredCapability(imageIntent, platform)
}
func allowOpenAICompatibleMessagesDispatch(apiKey *service.APIKey) bool {
if apiKey == nil || apiKey.Group == nil {
return true
@@ -288,6 +298,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
if !ok {
return
}
legacyCompact := service.IsOpenAIResponsesCompactPath(c)
nativeV2 := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(body)
// body-signal compact:上游 unary 等待期间向下游发 SSE 注释行心跳,防止
// 反向代理空闲超时掐断长压缩连接(#3887)。首拍延迟一个心跳间隔,快速
// 失败仍走 JSON+状态码链路;未标记客户端流式或间隔为 0 时是 no-op。
@@ -377,6 +389,15 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
channelMapping, _ := h.gatewayService.ResolveChannelMappingAndRestrict(c.Request.Context(), apiKey.GroupID, reqModel)
forwardBody := openAIModelMappedBody(body, channelMapping.Mapped, channelMapping.MappedModel, h.gatewayService.ReplaceModelInBody)
seedOpenAIForwardImageIntentHint(c, channelMapping.Mapped, imageIntent)
forwardModel := reqModel
if channelMapping.Mapped {
forwardModel = channelMapping.MappedModel
}
c.Request = c.Request.WithContext(service.WithOpenAIForwardModel(
c.Request.Context(),
forwardModel,
legacyCompact,
))
// 提前校验 function_call_output 是否具备可关联上下文,避免上游 400。
if !h.validateFunctionCallOutputRequest(c, body, reqLog) {
@@ -420,7 +441,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
if h.rejectIfCyberSessionBlocked(c, apiKey, sessionHashBody, reqModel, cyberBlockFormatResponses) {
return
}
requireCompact := isOpenAIRemoteCompactPath(c)
requireCompact := legacyCompact
maxAccountSwitches := h.maxAccountSwitches
switchCount := 0
@@ -437,7 +458,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
// 仅对 OpenAI 平台生效:Grok 生图走独立的 forwardGrokResponses 路径,不应被过滤。
// 复用前置权限与并发阶段在未修改 body 上确认的显式生图意图,避免大 tools 请求重复扫描。
// 该判断已排除 Codex 被动 image_gen namespace,避免 CC-only 账号被误过滤(#4476)。
requiredCapability := openAIResponsesRequiredCapability(imageIntent, requestPlatform)
needsResponses := nativeV2 || legacyCompact
requiredCapability := openAIResponsesRequiredCapabilityForRequest(imageIntent, needsResponses, requestPlatform)
// 分组利润控制:请求级装配定价上下文——pricingAt 固定本请求的
// D 与计费高峰因子,选号、槽位终检与全部 failover 重入共用同一门与阈值。
@@ -479,7 +501,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
zap.Int("excluded_account_count", len(failedAccountIDs)),
)
if len(failedAccountIDs) == 0 {
if errors.Is(err, service.ErrNoAvailableCompactAccounts) {
if legacyCompact && errors.Is(err, service.ErrNoAvailableCompactAccounts) {
markOpsRoutingCapacityLimitedIfNoAvailable(c, err)
h.handleStreamingAwareError(c, http.StatusServiceUnavailable, "compact_not_supported", "No available accounts support /responses/compact", streamStarted)
return
@@ -735,12 +757,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
}
}
func isOpenAIRemoteCompactPath(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/compact")
func isOpenAILegacyCompactPath(c *gin.Context) bool {
return service.IsOpenAIResponsesCompactPath(c)
}
// isBareOpenAIResponsesPath 仅匹配裸 /responses 端点(无 /compact 等子路径),
@@ -750,32 +768,27 @@ func isBareOpenAIResponsesPath(c *gin.Context) bool {
return false
}
normalizedPath := strings.TrimRight(strings.TrimSpace(c.Request.URL.Path), "/")
return strings.HasSuffix(normalizedPath, "/responses")
}
func isOpenAIRemoteCompactionV2Request(c *gin.Context, body []byte) bool {
stream, valid := parseOpenAICompatibleStream(body)
if !valid || !stream || c == nil || c.Request == nil {
switch normalizedPath {
case EndpointResponses, "/openai/v1/responses", "/responses", "/backend-api/codex/responses":
return true
default:
return false
}
for _, header := range c.Request.Header.Values("x-codex-beta-features") {
for _, feature := range strings.Split(header, ",") {
if strings.TrimSpace(feature) == "remote_compaction_v2" {
return true
}
}
}
return false
}
func isOpenAIRemoteCompactionV2Request(body []byte) bool {
stream, valid := parseOpenAICompatibleStream(body)
return valid && stream && service.HasCompactionTriggerInInput(body)
}
// normalizeOpenAIResponsesCompactRequest keeps Codex remote compaction v2 on
// its native streaming /responses wire and preserves the legacy body-signal
// promotion for clients that do not explicitly advertise that protocol.
// promotion for non-streaming requests.
// 返回归一化后的 body;ok=false 表示错误响应已写出,调用方应直接 return。
func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Context, reqLog *zap.Logger, body []byte) ([]byte, bool) {
isCompactRequest := service.IsOpenAIResponsesCompactPathForTest(c)
isCompactRequest := isOpenAILegacyCompactPath(c)
if !isCompactRequest && isBareOpenAIResponsesPath(c) && service.HasCompactionTriggerInInput(body) {
if isOpenAIRemoteCompactionV2Request(c, body) {
if isOpenAIRemoteCompactionV2Request(body) {
return body, true
}
c.Request.URL.Path = strings.TrimRight(c.Request.URL.Path, "/") + "/compact"
@@ -804,7 +817,7 @@ func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Con
}
func (h *OpenAIGatewayHandler) logOpenAIRemoteCompactOutcome(c *gin.Context, startedAt time.Time) {
if !isOpenAIRemoteCompactPath(c) {
if !isOpenAILegacyCompactPath(c) {
return
}
@@ -5,8 +5,7 @@ package handler
// 槽位终检与生图跳门回归(handler 半程):
// - 槽位获取成功后的利润终检:越线账号释放槽位并要求调用方排除重选,
// 不写响应、不绑定粘连;
// - openAIResponsesRequiredCapability 的生图意图映射钉死(scheduler 的
// 跳门条件依赖 CapabilityResponses ⇔ 显式生图意图这一耦合)。
// - openAIResponsesRequiredCapability 的请求能力映射覆盖生图与原生远程压缩。
import (
"context"
@@ -141,10 +140,10 @@ func TestAcquireResponsesAccountSlotProfitRecheck(t *testing.T) {
})
}
// scheduler 跳门条件依赖"CapabilityResponses 仅在显式生图意图时被要求"这一
// 映射;后续若扩展该 capability 的用途,本测试失败提示同步收窄跳门条件。
func TestOpenAIResponsesRequiredCapabilityPinsImageIntentMapping(t *testing.T) {
func TestOpenAIResponsesRequiredCapabilityForRequest(t *testing.T) {
require.Equal(t, service.OpenAIEndpointCapabilityResponses, openAIResponsesRequiredCapability(true, service.PlatformOpenAI))
require.Equal(t, service.OpenAIEndpointCapabilityChatCompletions, openAIResponsesRequiredCapability(false, service.PlatformOpenAI))
require.Equal(t, service.OpenAIEndpointCapabilityChatCompletions, openAIResponsesRequiredCapability(true, service.PlatformGrok))
require.Equal(t, service.OpenAIEndpointCapabilityResponses, openAIResponsesRequiredCapabilityForRequest(false, true, service.PlatformOpenAI))
require.Equal(t, service.OpenAIEndpointCapabilityChatCompletions, openAIResponsesRequiredCapabilityForRequest(false, true, service.PlatformGrok))
}
@@ -82,8 +82,10 @@ type OpenAIAccountScheduleRequest struct {
RequiredTransport OpenAIUpstreamTransport
RequiredCapability OpenAIEndpointCapability
RequiredImageCapability OpenAIImagesCapability
RequireCompact bool
ExcludedIDs map[int64]struct{}
// RequireCompact is only for legacy /responses/compact capability filtering
// and compact_model_mapping; native remote compaction v2 leaves it false.
RequireCompact bool
ExcludedIDs map[int64]struct{}
}
type OpenAIAccountScheduleDecision struct {
@@ -2154,11 +2156,9 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
// 入口已在请求开始经 WithOpenAIRequestPricingContext 装门并固定 pricingAt,
// 此处对同分组门直接复用(failover 重入阈值稳定),仅为不经 handler 装配的
// 内部调用兜底。图片/视频调度不在利润门范围:requiredImageCapability 非空的
// Images 调度不装门;requiredCapability == OpenAIEndpointCapabilityResponses
// 当前仅显式生图意图的 /v1/responses 设置(HTTP openAIResponsesRequiredCapability
// 与 WS 桥同款判定),同样不装门——若未来把该 capability 用于非生图流量,
// 需要同步收窄本条件(有测试钉死该映射)。
if requiredImageCapability == "" && requiredCapability != OpenAIEndpointCapabilityResponses {
// Images 调度不装门;其他使用 Responses 能力的文本请求(包括原生远程压缩)
// 仍须装门。其余媒体路径通过 WithOpenAIProfitControlSuppressed 显式跳过。
if requiredImageCapability == "" {
ctx = s.withOpenAIProfitControlGate(ctx, groupID)
}
platform = normalizeOpenAICompatiblePlatform(platform)
@@ -116,6 +116,262 @@ func TestOpenAIGatewayService_SelectAccountWithScheduler_CompactRejectsExplicitl
require.Nil(t, selection)
}
func newOpenAICompactionSchedulerTestService(accounts []Account, advanced bool) *OpenAIGatewayService {
cfg := &config.Config{}
cfg.Gateway.Scheduling.LoadBatchEnabled = false
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: &schedulerTestGatewayCache{},
cfg: cfg,
concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{}),
}
if advanced {
svc.rateLimitService = newOpenAIAdvancedSchedulerRateLimitService("true")
}
return svc
}
func selectOpenAICompactionSchedulerTestAccount(t *testing.T, svc *OpenAIGatewayService, groupID int64, requireCompact bool) (*AccountSelectionResult, error) {
t.Helper()
selection, _, err := svc.SelectAccountWithSchedulerForCapability(
context.Background(),
&groupID,
"",
"",
"gpt-5.6-sol",
nil,
OpenAIUpstreamTransportAny,
OpenAIEndpointCapabilityResponses,
requireCompact,
false,
false,
)
return selection, err
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_NativeCompactionIgnoresLegacyCompactProbe(t *testing.T) {
for _, advanced := range []bool{false, true} {
t.Run(map[bool]string{false: "legacy_scheduler", true: "advanced_scheduler"}[advanced], func(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
svc := newOpenAICompactionSchedulerTestService([]Account{{
ID: 71012,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_supported": false,
"openai_responses_supported": true,
},
}}, advanced)
selection, err := selectOpenAICompactionSchedulerTestAccount(t, svc, 91007, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.Equal(t, int64(71012), selection.Account.ID)
})
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_NativeCompactionAllowsForceOff(t *testing.T) {
for _, advanced := range []bool{false, true} {
t.Run(map[bool]string{false: "legacy_scheduler", true: "advanced_scheduler"}[advanced], func(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
svc := newOpenAICompactionSchedulerTestService([]Account{{
ID: 71013,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_mode": OpenAICompactModeForceOff,
"openai_responses_supported": true,
},
}}, advanced)
selection, err := selectOpenAICompactionSchedulerTestAccount(t, svc, 91008, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.Equal(t, int64(71013), selection.Account.ID)
})
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_NativeCompactionRequiresResponsesCapability(t *testing.T) {
for _, advanced := range []bool{false, true} {
t.Run(map[bool]string{false: "legacy_scheduler", true: "advanced_scheduler"}[advanced], func(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
svc := newOpenAICompactionSchedulerTestService([]Account{{
ID: 71014,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_mode": OpenAICompactModeForceOn,
"openai_responses_supported": false,
},
}}, advanced)
selection, err := selectOpenAICompactionSchedulerTestAccount(t, svc, 91009, false)
require.ErrorIs(t, err, ErrNoAvailableAccounts)
require.NotErrorIs(t, err, ErrNoAvailableCompactAccounts)
require.NotContains(t, err.Error(), "/responses/compact")
require.Nil(t, selection)
})
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_LegacyCompactionKeepsCompactEligibility(t *testing.T) {
for _, advanced := range []bool{false, true} {
t.Run(map[bool]string{false: "legacy_scheduler", true: "advanced_scheduler"}[advanced], func(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
svc := newOpenAICompactionSchedulerTestService([]Account{
{
ID: 71015,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_supported": false,
"openai_responses_supported": true,
},
},
{
ID: 71016,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_mode": OpenAICompactModeForceOff,
"openai_responses_supported": true,
},
},
}, advanced)
selection, err := selectOpenAICompactionSchedulerTestAccount(t, svc, 91010, true)
require.ErrorIs(t, err, ErrNoAvailableCompactAccounts)
require.Contains(t, err.Error(), "/responses/compact")
require.Nil(t, selection)
})
}
}
// TestOpenAIGatewayService_SelectAccountWithScheduler_CompactRequiresResponsesCapability
// prevents a confirmed Chat Completions-only API key from silently dropping the
// compaction trigger through the Responses-to-Chat fallback.
func TestOpenAIGatewayService_SelectAccountWithScheduler_CompactRequiresResponsesCapability(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
ctx := context.Background()
groupID := int64(91005)
accounts := []Account{{
ID: 71050,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Extra: map[string]any{
"openai_compact_supported": true,
"openai_responses_supported": false,
},
}}
cfg := &config.Config{}
cfg.Gateway.Scheduling.LoadBatchEnabled = false
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: &schedulerTestGatewayCache{},
cfg: cfg,
concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{}),
}
selection, _, err := svc.SelectAccountWithSchedulerForCapability(
ctx,
&groupID,
"",
"",
"gpt-5.4",
nil,
OpenAIUpstreamTransportAny,
OpenAIEndpointCapabilityResponses,
true,
false,
false,
)
require.Error(t, err)
require.True(t, errors.Is(err, ErrNoAvailableAccounts))
require.Nil(t, selection)
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_CompactSkipsChatOnlyAccount(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
ctx := context.Background()
groupID := int64(91006)
accounts := []Account{
{
ID: 71060,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Priority: 10,
Extra: map[string]any{
"openai_compact_supported": true,
"openai_responses_supported": false,
},
},
{
ID: 71061,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Priority: 0,
Extra: map[string]any{
"openai_compact_supported": true,
"openai_responses_supported": true,
},
},
}
cfg := &config.Config{}
cfg.Gateway.Scheduling.LoadBatchEnabled = false
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: &schedulerTestGatewayCache{},
cfg: cfg,
concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{}),
}
selection, _, err := svc.SelectAccountWithSchedulerForCapability(
ctx,
&groupID,
"",
"",
"gpt-5.6-sol",
nil,
OpenAIUpstreamTransportAny,
OpenAIEndpointCapabilityResponses,
true,
false,
false,
)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(71061), selection.Account.ID)
}
// TestOpenAIGatewayService_SelectAccountWithScheduler_CompactFallsBackToUnknown
// 验证当没有"已知支持"账号时,compact 请求会回退到"未探测"账号。
func TestOpenAIGatewayService_SelectAccountWithScheduler_CompactFallsBackToUnknown(t *testing.T) {
@@ -86,6 +86,226 @@ func TestOpenAISelectAccountForModelWithExclusions_UpstreamRestrictionSkipsDisal
require.Equal(t, int64(2), account.ID)
}
func TestIsUpstreamModelRestrictedByChannel_CompactMappingMatchesForwardPath(t *testing.T) {
t.Parallel()
account := &Account{
Platform: PlatformOpenAI,
Credentials: map[string]any{
"model_mapping": map[string]any{"gpt-5.4-channel": "gpt-5.4-account"},
"compact_model_mapping": map[string]any{"gpt-5.4-account": "gpt-5.4-compact"},
},
}
tests := []struct {
name string
allowedUpstreamModel string
useCompactModelMapping bool
}{
{
name: "legacy compact applies compact mapping after channel and account mapping",
allowedUpstreamModel: "gpt-5.4-compact",
useCompactModelMapping: true,
},
{
name: "native v2 stops after channel and account mapping",
allowedUpstreamModel: "gpt-5.4-account",
useCompactModelMapping: false,
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
channelSvc := newTestChannelService(makeStandardRepo(Channel{
ID: 1,
Status: StatusActive,
GroupIDs: []int64{10},
RestrictModels: true,
BillingModelSource: BillingModelSourceUpstream,
ModelPricing: []ChannelModelPricing{
{Platform: PlatformOpenAI, Models: []string{tt.allowedUpstreamModel}},
},
ModelMapping: map[string]map[string]string{
PlatformOpenAI: {"gpt-5.4": "gpt-5.4-channel"},
},
}, map[int64]string{10: PlatformOpenAI}))
svc := &OpenAIGatewayService{channelService: channelSvc}
mapping := channelSvc.ResolveChannelMapping(context.Background(), 10, "gpt-5.4")
require.True(t, mapping.Mapped)
require.Equal(t, "gpt-5.4-channel", mapping.MappedModel)
ctx := WithOpenAIForwardModel(
context.Background(),
mapping.MappedModel,
tt.useCompactModelMapping,
)
require.False(t, svc.isUpstreamModelRestrictedByChannel(
ctx, 10, account, "gpt-5.4", true,
))
require.True(t, svc.isUpstreamModelRestrictedByChannel(
context.Background(), 10, account, "gpt-5.4", true,
), "without the forward-model context the restriction check follows a different chain")
})
}
}
func TestIsUpstreamModelRestrictedByChannel_PassthroughMatchesForwardPath(t *testing.T) {
t.Parallel()
account := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Credentials: map[string]any{
"model_mapping": map[string]any{"gpt-5.4-channel": "gpt-5.4-account"},
"compact_model_mapping": map[string]any{
"gpt-5.4-channel": "gpt-5.4-compact",
},
},
Extra: map[string]any{"openai_passthrough": true},
}
tests := []struct {
name string
allowedUpstreamModel string
useCompactModelMapping bool
}{
{
name: "native v2 keeps channel-mapped model and ignores normal account mapping",
allowedUpstreamModel: "gpt-5.4-channel",
useCompactModelMapping: false,
},
{
name: "legacy compact applies compact mapping to channel-mapped model",
allowedUpstreamModel: "gpt-5.4-compact",
useCompactModelMapping: true,
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
channelSvc := newTestChannelService(makeStandardRepo(Channel{
ID: 1,
Status: StatusActive,
GroupIDs: []int64{10},
RestrictModels: true,
BillingModelSource: BillingModelSourceUpstream,
ModelPricing: []ChannelModelPricing{
{Platform: PlatformOpenAI, Models: []string{tt.allowedUpstreamModel}},
},
ModelMapping: map[string]map[string]string{
PlatformOpenAI: {"gpt-5.4": "gpt-5.4-channel"},
},
}, map[int64]string{10: PlatformOpenAI}))
svc := &OpenAIGatewayService{channelService: channelSvc}
mapping := channelSvc.ResolveChannelMapping(context.Background(), 10, "gpt-5.4")
require.True(t, mapping.Mapped)
require.Equal(t, "gpt-5.4-channel", mapping.MappedModel)
ctx := WithOpenAIForwardModel(
context.Background(),
mapping.MappedModel,
tt.useCompactModelMapping,
)
require.False(t, svc.isUpstreamModelRestrictedByChannel(
ctx, 10, account, "gpt-5.4", true,
))
})
}
}
func TestIsUpstreamModelRestrictedByChannel_PassthroughFlagWithRawChatFallbackMatchesForwardPath(t *testing.T) {
t.Parallel()
account := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Credentials: map[string]any{
"model_mapping": map[string]any{"gpt-5.4-channel": "gpt-5.4-account"},
"compact_model_mapping": map[string]any{
"gpt-5.4-account": "gpt-5.4-compact",
},
},
Extra: map[string]any{
"openai_passthrough": true,
"openai_responses_supported": false,
},
}
for _, useCompactModelMapping := range []bool{false, true} {
useCompactModelMapping := useCompactModelMapping
name := "native v2"
if useCompactModelMapping {
name = "legacy compact"
}
t.Run(name, func(t *testing.T) {
t.Parallel()
channelSvc := newTestChannelService(makeStandardRepo(Channel{
ID: 1,
Status: StatusActive,
GroupIDs: []int64{10},
RestrictModels: true,
BillingModelSource: BillingModelSourceUpstream,
ModelPricing: []ChannelModelPricing{
{Platform: PlatformOpenAI, Models: []string{"gpt-5.4-account"}},
},
ModelMapping: map[string]map[string]string{
PlatformOpenAI: {"gpt-5.4": "gpt-5.4-channel"},
},
}, map[int64]string{10: PlatformOpenAI}))
svc := &OpenAIGatewayService{channelService: channelSvc}
ctx := WithOpenAIForwardModel(
context.Background(),
"gpt-5.4-channel",
useCompactModelMapping,
)
require.False(t, svc.isUpstreamModelRestrictedByChannel(
ctx, 10, account, "gpt-5.4", true,
))
})
}
}
func TestIsUpstreamModelRestrictedByChannel_ForwardModelContextMatchesNormalForwardPath(t *testing.T) {
t.Parallel()
account := &Account{
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Credentials: map[string]any{
"model_mapping": map[string]any{"gpt-5.4-channel": "gpt-5.4-account"},
},
Extra: map[string]any{
"openai_passthrough": true,
"openai_responses_supported": false,
},
}
channelSvc := newTestChannelService(makeStandardRepo(Channel{
ID: 1,
Status: StatusActive,
GroupIDs: []int64{10},
RestrictModels: true,
BillingModelSource: BillingModelSourceUpstream,
ModelPricing: []ChannelModelPricing{
{Platform: PlatformOpenAI, Models: []string{"gpt-5.4-account"}},
},
ModelMapping: map[string]map[string]string{
PlatformOpenAI: {"gpt-5.4": "gpt-5.4-channel"},
},
}, map[int64]string{10: PlatformOpenAI}))
svc := &OpenAIGatewayService{channelService: channelSvc}
ctx := WithOpenAIForwardModel(context.Background(), "gpt-5.4-channel", false)
require.False(t, svc.isUpstreamModelRestrictedByChannel(
ctx, 10, account, "gpt-5.4", false,
))
}
func TestOpenAISelectAccountForModelWithExclusions_StickyRestrictedUpstreamFallsBack(t *testing.T) {
t.Parallel()
@@ -4,8 +4,8 @@ import "github.com/tidwall/gjson"
// HasCompactionTriggerInInput detects an input item with
// type="compaction_trigger". The handler combines this body signal with the
// request path, stream flag, and Codex beta feature header to distinguish the
// native remote compaction v2 wire from the legacy /responses/compact bridge.
// request path and stream flag to distinguish the native remote compaction v2
// wire from the legacy /responses/compact bridge.
func HasCompactionTriggerInInput(body []byte) bool {
if len(body) == 0 {
return false
@@ -0,0 +1,32 @@
package service
import "context"
type openAIForwardModelContextKey struct{}
type openAIForwardModel struct {
model string
useCompactModelMapping bool
}
// WithOpenAIForwardModel records the model present in the forwarded request
// body after channel mapping and whether the legacy /responses/compact-only
// model mapping applies. Native remote compaction v2 keeps this false, so
// channel restriction checks follow the same model chain used by Forward.
func WithOpenAIForwardModel(ctx context.Context, forwardModel string, useCompactModelMapping bool) context.Context {
if ctx == nil {
ctx = context.Background()
}
return context.WithValue(ctx, openAIForwardModelContextKey{}, openAIForwardModel{
model: forwardModel,
useCompactModelMapping: useCompactModelMapping,
})
}
func openAIForwardModelFromContext(ctx context.Context) (openAIForwardModel, bool) {
if ctx == nil {
return openAIForwardModel{}, false
}
forwardModel, ok := ctx.Value(openAIForwardModelContextKey{}).(openAIForwardModel)
return forwardModel, ok
}
@@ -108,7 +108,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
return s.forwardGrokResponses(ctx, c, account, body, originalModel, reqStream, startTime)
}
if account.Type == AccountTypeAPIKey && !openai_compat.ShouldUseResponsesAPI(account.Extra) {
if shouldForwardOpenAIResponsesViaRawChatCompletions(account) {
return s.forwardResponsesViaRawChatCompletions(ctx, c, account, body)
}
if account.Platform == PlatformOpenAI && account.Type == AccountTypeAPIKey {
@@ -1028,6 +1028,12 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
}
}
func shouldForwardOpenAIResponsesViaRawChatCompletions(account *Account) bool {
return account != nil &&
account.Type == AccountTypeAPIKey &&
!openai_compat.ShouldUseResponsesAPI(account.Extra)
}
func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.Context, account *Account, body []byte, token string, isStream bool, promptCacheKey string, isCodexCLI bool) (*http.Request, error) {
// Determine target URL based on account type
var targetURL string
@@ -276,7 +276,9 @@ func isOpenAIEncryptedReasoningInputItem(item any) bool {
return has
}
func IsOpenAIResponsesCompactPathForTest(c *gin.Context) bool {
// IsOpenAIResponsesCompactPath reports whether the request targets the legacy
// /responses/compact endpoint, including its forwardable subpaths.
func IsOpenAIResponsesCompactPath(c *gin.Context) bool {
return isOpenAIResponsesCompactPath(c)
}
@@ -53,6 +53,50 @@ func TestForwardResponses_ForceChatCompletionsRoutesNonStreamingToChatCompletion
require.False(t, result.Stream)
}
func TestForwardResponses_PassthroughFlagWithUnsupportedResponsesUsesAccountMapping(t *testing.T) {
gin.SetMode(gin.TestMode)
for _, path := range []string{"/v1/responses", "/v1/responses/compact"} {
path := path
t.Run(path, func(t *testing.T) {
body := []byte(`{"model":"gpt-5.4-channel","input":"hello","stream":false}`)
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")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(
`{"id":"chatcmpl_mapping","object":"chat.completion","model":"gpt-5.4-account","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`,
)),
}}
svc := &OpenAIGatewayService{
cfg: rawChatCompletionsTestConfig(),
httpUpstream: upstream,
}
account := rawChatCompletionsTestAccount()
account.Credentials["model_mapping"] = map[string]any{
"gpt-5.4-channel": "gpt-5.4-account",
}
account.Credentials["compact_model_mapping"] = map[string]any{
"gpt-5.4-account": "gpt-5.4-compact",
}
account.Extra = map[string]any{
"openai_passthrough": true,
openai_compat.ExtraKeyResponsesSupported: false,
}
result, err := svc.Forward(context.Background(), c, account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, "http://upstream.example/v1/chat/completions", upstream.lastReq.URL.String())
require.Equal(t, "gpt-5.4-account", gjson.GetBytes(upstream.lastBody, "model").String())
})
}
}
func TestForwardResponses_ForceChatCompletionsRoutesStreamingToChatCompletions(t *testing.T) {
gin.SetMode(gin.TestMode)
@@ -244,7 +244,7 @@ func (s *OpenAIGatewayService) SelectAccountForModelWithExclusions(ctx context.C
}
// noAvailableOpenAISelectionError builds the standard "no account available" error
// while preserving the compact-specific error when applicable.
// while preserving the legacy /responses/compact error when applicable.
func normalizeOpenAICompatiblePlatform(platform string) string {
if platform == PlatformGrok {
return PlatformGrok
@@ -685,17 +685,44 @@ func prioritizeOpenAICompactAccounts(accounts []*Account) []*Account {
}
// resolveOpenAIAccountUpstreamModelForRequest resolves the upstream model that
// would be sent for a given request, honouring compact-only mappings when the
// caller is on the /responses/compact path.
// would be sent for a given request, honoring the legacy compact-only mapping
// when the caller is on the /responses/compact path.
func resolveOpenAIAccountUpstreamModelForRequest(account *Account, requestedModel string, requireCompact bool) string {
// Forward checks the raw Chat Completions fallback before passthrough.
// These API-key accounts therefore apply normal account model_mapping and
// upstream normalization, but never compact_model_mapping.
if shouldForwardOpenAIResponsesViaRawChatCompletions(account) {
upstreamModel := resolveOpenAIForwardModel(account, requestedModel, "")
return normalizeOpenAIModelForUpstream(account, upstreamModel)
}
// Passthrough accounts only replace authentication. Their Forward path
// keeps the channel-mapped model in the request body and does not apply the
// account's normal model_mapping. Legacy /responses/compact is the one
// exception: forwardOpenAIPassthrough applies compact_model_mapping
// directly to that channel-mapped model.
if account != nil && account.IsOpenAIPassthroughEnabled() {
upstreamModel := strings.TrimSpace(requestedModel)
if upstreamModel == "" {
return ""
}
if requireCompact {
return resolveOpenAICompactForwardModel(account, upstreamModel)
}
return upstreamModel
}
upstreamModel := resolveOpenAIForwardModel(account, requestedModel, "")
if upstreamModel == "" {
return ""
}
if requireCompact {
return resolveOpenAICompactForwardModel(account, upstreamModel)
compactModel := resolveOpenAICompactForwardModel(account, upstreamModel)
if compactModel != upstreamModel {
return compactModel
}
}
return upstreamModel
return normalizeOpenAIModelForUpstream(account, upstreamModel)
}
func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, error) {
@@ -815,7 +842,8 @@ func (s *OpenAIGatewayService) tryStickySessionHit(ctx context.Context, groupID
// selectBestAccount selects the best account from candidates (priority + LRU).
// Returns nil if no available account. The second return reports whether at
// least one candidate was filtered out solely because it lacks compact support
// (only meaningful when requireCompact=true); the third contains deterministic
// (only meaningful when the legacy /responses/compact requireCompact flag is
// true); the third contains deterministic
// exclusion diagnostics for the evaluated snapshot.
func (s *OpenAIGatewayService) selectBestAccount(ctx context.Context, groupID *int64, platform string, accounts []Account, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, bool, openAISelectionFilterStats) {
platform = normalizeOpenAICompatiblePlatform(platform)
@@ -398,8 +398,8 @@ func (t *accountWriteThrottle) Allow(id int64, now time.Time) bool {
var defaultOpenAICodexSnapshotPersistThrottle = newAccountWriteThrottle(openAICodexSnapshotPersistMinInterval)
// ErrNoAvailableCompactAccounts indicates the request needs /responses/compact
// support but no compatible account is available.
// ErrNoAvailableCompactAccounts indicates a legacy /responses/compact request
// needs compact support but no compatible account is available.
var ErrNoAvailableCompactAccounts = errors.New("no available accounts support /responses/compact")
// OpenAIGatewayService handles OpenAI API gateway operations
@@ -598,6 +598,10 @@ func (s *OpenAIGatewayService) isUpstreamModelRestrictedByChannel(ctx context.Co
if s.channelService == nil {
return false
}
if compactForwardModel, ok := openAIForwardModelFromContext(ctx); ok {
requestedModel = compactForwardModel.model
requireCompact = compactForwardModel.useCompactModelMapping
}
upstreamModel := resolveOpenAIAccountUpstreamModelForRequest(account, requestedModel, requireCompact)
if upstreamModel == "" {
return false
@@ -735,12 +735,17 @@ func TestOpenAIGatewayService_OAuthPassthrough_CompactUsesJSONAndKeepsNonStreami
}
account := &Account{
ID: 123,
Name: "acc",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"},
ID: 123,
Name: "acc",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{
"access_token": "oauth-token",
"chatgpt_account_id": "chatgpt-acc",
"model_mapping": map[string]any{"gpt-5.1-codex": "gpt-5.1-account"},
"compact_model_mapping": map[string]any{"gpt-5.1-codex": "gpt-5.1-compact"},
},
Extra: map[string]any{"openai_passthrough": true},
Status: StatusActive,
Schedulable: true,
@@ -754,7 +759,7 @@ func TestOpenAIGatewayService_OAuthPassthrough_CompactUsesJSONAndKeepsNonStreami
require.False(t, gjson.GetBytes(upstream.lastBody, "store").Exists())
require.False(t, gjson.GetBytes(upstream.lastBody, "stream").Exists())
require.Equal(t, "gpt-5.1-codex", gjson.GetBytes(upstream.lastBody, "model").String())
require.Equal(t, "gpt-5.1-compact", gjson.GetBytes(upstream.lastBody, "model").String())
require.Equal(t, "compact me", gjson.GetBytes(upstream.lastBody, "input.0.text").String())
require.Equal(t, "local-test-instructions", strings.TrimSpace(gjson.GetBytes(upstream.lastBody, "instructions").String()))
require.Equal(t, "application/json", upstream.lastReq.Header.Get("Accept"))
@@ -2172,12 +2177,16 @@ func TestOpenAIGatewayService_APIKeyPassthrough_PreservesBodyAndUsesResponsesEnd
}
account := &Account{
ID: 456,
Name: "apikey-acc",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "sk-api-key", "base_url": "https://api.openai.com"},
ID: 456,
Name: "apikey-acc",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{
"api_key": "sk-api-key",
"base_url": "https://api.openai.com",
"model_mapping": map[string]any{"gpt-5.2": "gpt-5.2-account"},
},
Extra: map[string]any{"openai_passthrough": true},
Status: StatusActive,
Schedulable: true,
@@ -1,7 +1,7 @@
package service
// 请求级定价与利润门回归:请求级 pricingAt 定价上下文、门复用(failover 阈值稳定)、
// 生图意图跳门、U 使用账号倍率且与探测新鲜度解耦、
// Responses 文本能力利润门、U 使用账号倍率且与探测新鲜度解耦、
// 用量记录定价时刻取值。
import (
@@ -109,8 +109,9 @@ func TestProfitControl_UsesAccountRateInsteadOfProbeSnapshot(t *testing.T) {
require.Equal(t, openAIProfitFilterReasonThreshold, reason)
}
// 显式生图意图(requiredCapability=Responses)在唯一调度入口跳门(图片边界不装门)。
func TestProfitControl_ResponsesImageIntentSkipsGateAtScheduler(t *testing.T) {
// Responses 是端点能力,不代表媒体请求;原生远程压缩同样要求该能力,
// 因此唯一文本调度入口必须照常安装利润门。
func TestProfitControl_ResponsesCapabilityUsesTextGateAtScheduler(t *testing.T) {
now := time.Now()
expensive := upstreamCostTestAccount(51, UpstreamBillingProbeStatusOK, 0.8, now.Add(-time.Minute), 30*time.Minute)
expensive.Status = StatusActive
@@ -129,12 +130,8 @@ func TestProfitControl_ResponsesImageIntentSkipsGateAtScheduler(t *testing.T) {
require.ErrorIs(t, err, ErrNoAvailableAccounts, "文本能力必须过利润门")
selection, _, err := svc.SelectAccountWithSchedulerForCapability(ctx, &groupID, "", "", "gpt-test", nil, OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityResponses, false, false, true)
require.NoError(t, err, "生图意图不装门,保持既有调度")
require.NotNil(t, selection)
require.Equal(t, expensive.ID, selection.Account.ID)
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
require.ErrorIs(t, err, ErrNoAvailableAccounts, "Responses 文本能力不得绕过利润门")
require.Nil(t, selection)
}
// 账号倍率缺失一律视为非法保守拒绝;手工或同步维护了倍率的任意账号类型都按
@@ -145,6 +145,34 @@ func TestOpenAIResponsesRequestPathSuffixRejectsNonConformingSubpaths(t *testing
}
}
func TestIsOpenAIResponsesCompactPathUsesLegacyEndpointShape(t *testing.T) {
legacyPaths := []string{
"/v1/responses/compact",
"/v1/responses/compact/detail",
"/responses/compact/",
}
for _, path := range legacyPaths {
t.Run("legacy_"+path, func(t *testing.T) {
c := newResponsesSuffixTestContext(t, path)
require.True(t, IsOpenAIResponsesCompactPath(c))
})
}
nonLegacyPaths := []string{
"/v1/responses",
"/openai/v1/responses",
"/responses",
"/backend-api/codex/responses",
"/v1/responses/resp_123/cancel",
}
for _, path := range nonLegacyPaths {
t.Run("non_legacy_"+path, func(t *testing.T) {
c := newResponsesSuffixTestContext(t, path)
require.False(t, IsOpenAIResponsesCompactPath(c))
})
}
}
func TestAppendOpenAIResponsesRequestPathSuffixRefusesUnsafeSuffix(t *testing.T) {
// 调用方漏了校验时,拼接函数本身也不得把不合规片段带进上游 URL。
require.Equal(t, chatgptCodexURL, appendOpenAIResponsesRequestPathSuffix(chatgptCodexURL, "/../../x"))