From 8ae6d8f67e72b099ed581b1455840ca62bb25561 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 15 Aug 2026 16:35:08 +0800 Subject: [PATCH] fix(openai): send session-level beta features and probe native compaction v2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenAI sunset the legacy unary /responses/compact endpoint (404, #5598, #5624), so the account "compact probe" in the admin UI kept failing even for healthy accounts, and the beta-feature negotiation header was only attached to compaction turns. Beta features (codex-rs session/mod.rs build_model_client_beta_features_header + client.rs build_responses_headers): the header is a session-level constant attached to every /responses request, the WS handshake and /responses/compact. Enumerating FEATURES shows no Experimental feature is enabled by default, so a default install sends exactly "remote_compaction_v2". Mirror that: - OAuth requests without a client-declared header get the default shape, so we no longer produce a "header only on compaction turns" pattern real Codex never emits (#5586 chains that strip the header) - a client-declared header is preserved as-is: non-empty without v2 means the user disabled the feature and the gateway must not rewrite that - native v2 turns (compaction_trigger in body) always ensure v2 is present - non-OAuth upstreams keep the compaction-turn-only behaviour - the WS injection sits outside the client-header copy block so prewarm and turn handshakes cannot land in different pool compatibility buckets Compact probe now exercises native v2 (streaming /responses + compaction_trigger) instead of the dead endpoint. Success requires an actual compaction output item — scanning output_item.done/added, the terminal response.output[] and the whole-JSON fallback — so a 2xx that silently drops the trigger is reported as unsupported (the "got 0 items" class, #5478, #5648). Probe identity is now UUID-shaped and applies the account's convergence, matching real traffic on the same endpoint. --- .../handler/openai_gateway_handler.go | 5 + .../internal/service/account_test_service.go | 42 +++-- ...ccount_test_service_openai_compact_test.go | 154 ++++++++++++++++-- .../openai_agent_identity_compat_test.go | 4 +- .../service/openai_compact_body_signal.go | 110 ++++++++++++- .../internal/service/openai_compact_probe.go | 62 ++++++- .../service/openai_compact_probe_test.go | 90 +++++++++- .../service/openai_ws_forwarder_payload.go | 6 + 8 files changed, 428 insertions(+), 45 deletions(-) diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 19373b4c41..1695eccc1b 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -300,6 +300,11 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { } legacyCompact := service.IsOpenAIResponsesCompactPath(c) nativeV2 := isBareOpenAIResponsesPath(c) && isOpenAIRemoteCompactionV2Request(body) + if nativeV2 { + // 原生 v2 压缩出站前补注 x-codex-beta-features: remote_compaction_v2， + // 与真实 Codex 线型一致（网关链剥头后本级负责恢复，#5586）。 + service.MarkOpenAINativeCompactionV2(c) + } // body-signal compact：上游 unary 等待期间向下游发 SSE 注释行心跳，防止 // 反向代理空闲超时掐断长压缩连接（#3887）。首拍延迟一个心跳间隔，快速 // 失败仍走 JSON+状态码链路；未标记客户端流式或间隔为 0 时是 no-op。 diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index 9e1d29c2f9..496ef57aed 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -610,10 +610,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account } // Align test routing with gateway behavior: OpenAI accounts apply normal - // account model mapping, and compact mode applies compact-only mapping on top. + // account model mapping. Native remote compaction v2 rides the ordinary + // /responses wire and does NOT apply the legacy compact-only mapping + // (post-#5641 semantics: compact_model_mapping is /responses/compact-only). testModelID = account.GetMappedModel(testModelID) if mode == AccountTestModeCompact { - testModelID = resolveOpenAICompactForwardModel(account, testModelID) return s.testOpenAICompactConnection(c, account, testModelID) } @@ -1976,8 +1977,10 @@ func (s *AccountTestService) testOpenAIChatCompletionsConnection( return s.processOpenAIChatCompletionsStream(c, resp.Body) } -// testOpenAICompactConnection probes /responses/compact and persists the -// resulting capability state on the account. +// testOpenAICompactConnection probes native remote compaction v2 (streaming +// /responses with a compaction_trigger input item) and persists the resulting +// capability state on the account. The legacy unary /responses/compact +// endpoint has been sunset upstream (404, #5598/#5624) and is no longer probed. func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account *Account, testModelID string) error { ctx := c.Request.Context() credentialAccount := account @@ -2002,7 +2005,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() { return s.sendErrorAndEnd(c, "No access token available") } - apiURL = chatgptCodexAPIURL + "/compact" + apiURL = chatgptCodexAPIURL case account.Type == AccountTypeAPIKey: authToken = account.GetOpenAIApiKey() if authToken == "" { @@ -2016,7 +2019,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid base URL: %s", err.Error())) } - apiURL = appendOpenAIResponsesRequestPathSuffix(buildOpenAIResponsesURL(normalizedBaseURL), "/compact") + apiURL = buildOpenAIResponsesURL(normalizedBaseURL) default: return s.sendErrorAndEnd(c, fmt.Sprintf("Unsupported account type: %s", account.Type)) } @@ -2027,7 +2030,11 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account c.Writer.Header().Set("X-Accel-Buffering", "no") c.Writer.Flush() - payloadBytes, _ := json.Marshal(createOpenAICompactProbePayload(testModelID)) + // 原生 v2 走普通 /responses 线：OAuth 与真实转发一致做上游模型归一化。 + if isOAuth { + testModelID = normalizeOpenAIModelForUpstream(credentialAccount, testModelID) + } + payloadBytes, _ := json.Marshal(createOpenAICompactProbePayload(testModelID, isOAuth)) if !agentIdentityTaskRecoveryWasTried(ctx) { s.sendEvent(c, TestEvent{Type: "test_start", Model: testModelID}) } @@ -2039,7 +2046,9 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) req.Header.Set("Content-Type", "application/json") - req.Header.Set("Accept", "application/json") + // v2 探测是流式请求；同时补注协商头，与真实 codex 出站线型一致。 + req.Header.Set("Accept", "text/event-stream") + ensureOpenAIRemoteCompactionV2BetaFeature(req.Header) if credentialAccount.IsOpenAIAgentIdentity() { authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.agentIdentityWS, &s.agentIdentityTaskMu, credentialAccount) if authErr != nil { @@ -2061,6 +2070,12 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account if isOAuth { req.Host = "chatgpt.com" setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount) + // 指纹收敛：探测与真实转发走同一个 /responses 端点，身份也必须同构， + // 否则探测流量会以「缺 x-codex-installation-id + 非收敛 session」的 + // 形态暴露在上游眼里。账号关闭收敛（off）时返回 nil，探测保持原样。 + if fpIDs := resolveCodexFingerprintIDsFromRequest(account, req.Header); fpIDs != nil { + applyCodexFingerprintHeaders(req.Header, fpIDs) + } } // 账号级请求头覆写：测试请求与真实转发保持一致的最终头 @@ -2074,7 +2089,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account resp, err := s.httpUpstream.DoWithTLS(req, proxyURL, account.ID, account.Concurrency, s.tlsFPProfileService.ResolveTLSProfile(account)) if err != nil { if s.accountRepo != nil { - updates := buildOpenAICompactProbeExtraUpdates(nil, nil, err, time.Now()) + updates := buildOpenAICompactProbeExtraUpdates(nil, nil, err, false, time.Now()) _ = s.accountRepo.UpdateExtra(ctx, account.ID, updates) mergeAccountExtra(account, updates) } @@ -2093,8 +2108,9 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account return s.testOpenAICompactConnection(c, account, testModelID) } + compactionFound := openAICompactProbeFoundCompactionItem(body) if s.accountRepo != nil { - updates := buildOpenAICompactProbeExtraUpdates(resp, body, nil, time.Now()) + updates := buildOpenAICompactProbeExtraUpdates(resp, body, nil, compactionFound, time.Now()) if codexUpdates, err := extractOpenAICodexProbeUpdates(resp); err == nil && len(codexUpdates) > 0 { updates = mergeExtraUpdates(updates, codexUpdates) } @@ -2116,7 +2132,11 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account return s.sendErrorAndEnd(c, fmt.Sprintf("API returned %d: %s", resp.StatusCode, string(body))) } - s.sendEvent(c, TestEvent{Type: "content", Text: "Compact probe succeeded"}) + if !compactionFound { + return s.sendErrorAndEnd(c, "Upstream returned 2xx without a compaction output item (native remote compaction v2 unsupported on this chain)") + } + + s.sendEvent(c, TestEvent{Type: "content", Text: "Compact probe succeeded (native remote compaction v2)"}) s.sendEvent(c, TestEvent{Type: "test_complete", Success: true}) return nil } diff --git a/backend/internal/service/account_test_service_openai_compact_test.go b/backend/internal/service/account_test_service_openai_compact_test.go index 5b973285a8..5b47725084 100644 --- a/backend/internal/service/account_test_service_openai_compact_test.go +++ b/backend/internal/service/account_test_service_openai_compact_test.go @@ -10,10 +10,16 @@ import ( "github.com/Wei-Shaw/sub2api/internal/config" "github.com/gin-gonic/gin" + "github.com/google/uuid" "github.com/stretchr/testify/require" "github.com/tidwall/gjson" ) +// compactProbeSSESuccessBody 是原生 v2 压缩成功的最小 SSE 形态： +// output_item.done 携带 compaction item + response.completed。 +const compactProbeSSESuccessBody = "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"compaction\",\"id\":\"cmp_probe\",\"encrypted_content\":\"blob\"}}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_probe\",\"output\":[]}}\n\n" + func TestAccountTestService_TestAccountConnection_OpenAICompactOAuthSuccessPersistsSupport(t *testing.T) { gin.SetMode(gin.TestMode) @@ -38,8 +44,8 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactOAuthSuccessPersi } upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid-probe"}}, - Body: io.NopCloser(strings.NewReader(`{"id":"cmp_probe","status":"completed"}`)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid-probe"}}, + Body: io.NopCloser(strings.NewReader(compactProbeSSESuccessBody)), }} svc := &AccountTestService{ accountRepo: repo, @@ -53,16 +59,22 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactOAuthSuccessPersi err := svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact) require.NoError(t, err) - require.Equal(t, chatgptCodexAPIURL+"/compact", upstream.lastReq.URL.String()) + // 原生 v2：探测普通 /responses 线，不再打已下线的 /responses/compact。 + require.Equal(t, chatgptCodexAPIURL, upstream.lastReq.URL.String()) require.Equal(t, "chatgpt.com", upstream.lastReq.Host) - require.Equal(t, "application/json", upstream.lastReq.Header.Get("Accept")) - require.Equal(t, codexCLIVersion, upstream.lastReq.Header.Get("Version")) + require.Equal(t, "text/event-stream", upstream.lastReq.Header.Get("Accept")) + require.Contains(t, upstream.lastReq.Header.Get("x-codex-beta-features"), "remote_compaction_v2") require.NotEmpty(t, upstream.lastReq.Header.Get("Session_Id")) require.Equal(t, HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileFromContext(upstream.lastReq.Context())) require.Equal(t, codexCLIUserAgent, upstream.lastReq.Header.Get("User-Agent")) require.Equal(t, "chatgpt-acc", upstream.lastReq.Header.Get("chatgpt-account-id")) require.Equal(t, "true", upstream.lastReq.Header.Get("x-openai-fedramp")) require.Equal(t, "gpt-5.4", gjson.GetBytes(upstream.lastBody, "model").String()) + require.True(t, gjson.GetBytes(upstream.lastBody, "stream").Bool()) + require.False(t, gjson.GetBytes(upstream.lastBody, "store").Bool()) + inputItems := gjson.GetBytes(upstream.lastBody, "input").Array() + require.NotEmpty(t, inputItems) + require.Equal(t, "compaction_trigger", inputItems[len(inputItems)-1].Get("type").String()) updates := <-updateCalls require.Equal(t, true, updates["openai_compact_supported"]) @@ -114,7 +126,7 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactOAuth404MarksUnsu require.Contains(t, rec.Body.String(), `"type":"error"`) } -func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyUsesCompactPath(t *testing.T) { +func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyUsesNativeResponsesPath(t *testing.T) { gin.SetMode(gin.TestMode) updateCalls := make(chan map[string]any, 1) @@ -127,8 +139,10 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyUsesCompact Schedulable: true, Concurrency: 1, Credentials: map[string]any{ - "api_key": "sk-test", - "base_url": "https://example.com/v1", + "api_key": "sk-test", + "base_url": "https://example.com/v1", + // post-#5641：compact_model_mapping 仅作用于 legacy /responses/compact， + // 原生 v2 探测不应用它。 "compact_model_mapping": map[string]any{"gpt-5.4": "gpt-5.4-openai-compact"}, }, } @@ -138,8 +152,8 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyUsesCompact } upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"application/json"}}, - Body: io.NopCloser(strings.NewReader(`{"id":"cmp_probe_apikey","status":"completed"}`)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(compactProbeSSESuccessBody)), }} svc := &AccountTestService{ accountRepo: repo, @@ -154,14 +168,16 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyUsesCompact err := svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact) require.NoError(t, err) - require.Equal(t, "https://example.com/v1/responses/compact", upstream.lastReq.URL.String()) + require.Equal(t, "https://example.com/v1/responses", upstream.lastReq.URL.String()) requireOpenAICodexProbeHeaders(t, upstream.lastReq.Header) - require.Equal(t, "gpt-5.4-openai-compact", gjson.GetBytes(upstream.lastBody, "model").String()) + require.Contains(t, upstream.lastReq.Header.Get("x-codex-beta-features"), "remote_compaction_v2") + require.Equal(t, "gpt-5.4", gjson.GetBytes(upstream.lastBody, "model").String(), + "原生 v2 探测不应用 compact_model_mapping") updates := <-updateCalls require.Equal(t, true, updates["openai_compact_supported"]) } -func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyDefaultBaseURLUsesV1Path(t *testing.T) { +func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyDefaultBaseURLUsesResponsesPath(t *testing.T) { gin.SetMode(gin.TestMode) updateCalls := make(chan map[string]any, 1) @@ -183,8 +199,8 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyDefaultBase } upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"application/json"}}, - Body: io.NopCloser(strings.NewReader(`{"id":"cmp_probe_apikey_default","status":"completed"}`)), + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(compactProbeSSESuccessBody)), }} svc := &AccountTestService{ accountRepo: repo, @@ -198,6 +214,112 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactAPIKeyDefaultBase err := svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact) require.NoError(t, err) - require.Equal(t, "https://api.openai.com/v1/responses/compact", upstream.lastReq.URL.String()) + require.Equal(t, "https://api.openai.com/v1/responses", upstream.lastReq.URL.String()) <-updateCalls } + +func TestAccountTestService_TestAccountConnection_OpenAICompact2xxWithoutItemMarksUnsupported(t *testing.T) { + gin.SetMode(gin.TestMode) + + updateCalls := make(chan map[string]any, 1) + account := Account{ + ID: 5, + Name: "openai-oauth-no-item", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{ + "access_token": "oauth-token", + "chatgpt_account_id": "chatgpt-acc", + }, + } + repo := &snapshotUpdateAccountRepo{ + stubOpenAIAccountRepo: stubOpenAIAccountRepo{accounts: []Account{account}}, + updateExtraCalls: updateCalls, + } + // 200 但流里没有 compaction item：链路吞掉了 compaction_trigger 的形态 + //（#5478 的 "got 0 items"），必须判定为不支持。 + noItemBody := "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_1\"}}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_x\",\"output\":[]}}\n\n" + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(noItemBody)), + }} + svc := &AccountTestService{ + accountRepo: repo, + httpUpstream: upstream, + } + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/5/test", bytes.NewReader(nil)) + + err := svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact) + require.Error(t, err) + + updates := <-updateCalls + require.Equal(t, false, updates["openai_compact_supported"]) + require.Contains(t, rec.Body.String(), `"type":"error"`) +} + +// 探测与真实转发走同一 /responses 端点，出站身份必须与真实 Codex 同构： +// session/thread 为 UUID、携带 x-codex-installation-id（收敛账号用收敛值）。 +func TestAccountTestService_TestAccountConnection_OpenAICompactProbeIdentityMatchesRealTraffic(t *testing.T) { + gin.SetMode(gin.TestMode) + + updateCalls := make(chan map[string]any, 1) + account := Account{ + ID: 6, + Name: "openai-oauth-identity", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{ + "access_token": "oauth-token", + "chatgpt_account_id": "chatgpt-acc", + }, + // 收敛是显式 opt-in（#5610），这里显式开启以验证探测身份与真实流量同构。 + Extra: map[string]any{"codex_fingerprint_mode": "session"}, + } + repo := &snapshotUpdateAccountRepo{ + stubOpenAIAccountRepo: stubOpenAIAccountRepo{accounts: []Account{account}}, + updateExtraCalls: updateCalls, + } + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(compactProbeSSESuccessBody)), + }} + svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/6/test", bytes.NewReader(nil)) + + require.NoError(t, svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact)) + + // 显式 session 收敛模式：出站身份 = 账号级收敛值 + converged := resolveConvergedSessionID(&account) + require.Equal(t, converged, upstream.lastReq.Header.Get("session-id")) + require.Equal(t, converged, upstream.lastReq.Header.Get("session_id")) + require.Equal(t, resolveConvergedInstallationID(&account), upstream.lastReq.Header.Get("x-codex-installation-id"), + "真实 Codex 每个请求必带 installation-id，探测不得缺失") + require.NotContains(t, upstream.lastReq.Header.Get("session-id"), "probe_compact", + "探测标识不得是可被上游一眼识别的字面量") + <-updateCalls +} + +func TestCompactProbeSessionID_IsUUIDShaped(t *testing.T) { + for _, id := range []int64{0, 1, 987654} { + got := compactProbeSessionID(id) + _, err := uuid.Parse(got) + require.NoError(t, err, "探测会话标识必须是 UUID 形态: %s", got) + } + require.Equal(t, compactProbeSessionID(7), compactProbeSessionID(7), "同账号应稳定复用同一会话") + require.NotEqual(t, compactProbeSessionID(7), compactProbeSessionID(8)) +} diff --git a/backend/internal/service/openai_agent_identity_compat_test.go b/backend/internal/service/openai_agent_identity_compat_test.go index 5dad35887b..a8f234f58c 100644 --- a/backend/internal/service/openai_agent_identity_compat_test.go +++ b/backend/internal/service/openai_agent_identity_compat_test.go @@ -40,7 +40,7 @@ func TestAccountTestServiceOpenAICompactAgentIdentityUsesFreshAssertion(t *testi upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, - Body: io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed"}`)), + Body: io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed","output":[{"type":"compaction","id":"cmp_agent_fresh","encrypted_content":"blob"}]}`)), }} svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream} @@ -87,7 +87,7 @@ func TestAccountTestServiceOpenAICompactAgentIdentityRecoversInvalidTaskOnce(t * upstream := &httpUpstreamRecorder{responses: []*http.Response{ {StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))}, - {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed"}`))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed","output":[{"type":"compaction","id":"cmp_agent","encrypted_content":"blob"}]}`))}, }} invalidator := &agentIdentityWSInvalidationRecorder{} svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream, agentIdentityWS: invalidator} diff --git a/backend/internal/service/openai_compact_body_signal.go b/backend/internal/service/openai_compact_body_signal.go index 5e7957c557..83266e6c00 100644 --- a/backend/internal/service/openai_compact_body_signal.go +++ b/backend/internal/service/openai_compact_body_signal.go @@ -1,6 +1,114 @@ package service -import "github.com/tidwall/gjson" +import ( + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/tidwall/gjson" +) + +// openAINativeCompactionV2Key 标记本请求是原生 remote compaction v2 +// （裸 /responses + stream:true + compaction_trigger），由 handler 在判定后 +// 写入，供上游请求构造时补注协商头。 +const openAINativeCompactionV2Key = "openai_native_compaction_v2" + +const openAIRemoteCompactionV2Feature = "remote_compaction_v2" + +// MarkOpenAINativeCompactionV2 由 handler 在识别出原生 v2 压缩请求时调用。 +func MarkOpenAINativeCompactionV2(c *gin.Context) { + if c != nil { + c.Set(openAINativeCompactionV2Key, true) + } +} + +func isOpenAINativeCompactionV2(c *gin.Context) bool { + if c == nil { + return false + } + return c.GetBool(openAINativeCompactionV2Key) +} + +// ensureOpenAIRemoteCompactionV2BetaFeature 确保出站 x-codex-beta-features +// 头包含 remote_compaction_v2。真实 Codex 发送 compaction_trigger 时总会同时 +// 携带该协商头（codex-rs build_model_client_beta_features_header 对该 feature +// 特判 advertise）；上游或下游网关链剥掉它后，请求会在依赖该头做门控的 +// 环节被降级（#5586）。这里在原生 v2 请求出站前补齐，使线型与真实 Codex +// 一致。已存在时保持原样，不重复追加。 +func ensureOpenAIRemoteCompactionV2BetaFeature(h http.Header) { + if h == nil { + return + } + tokens := make([]string, 0, 4) + for _, value := range h.Values("x-codex-beta-features") { + for _, token := range strings.Split(value, ",") { + token = strings.TrimSpace(token) + if token == "" { + continue + } + if token == openAIRemoteCompactionV2Feature { + return + } + tokens = append(tokens, token) + } + } + tokens = append(tokens, openAIRemoteCompactionV2Feature) + h.Set("x-codex-beta-features", strings.Join(tokens, ",")) +} + +// hasOpenAICodexBetaFeaturesHeader 报告出站头里是否已存在非空的 +// x-codex-beta-features（即客户端自己声明过能力集）。 +func hasOpenAICodexBetaFeaturesHeader(h http.Header) bool { + if h == nil { + return false + } + for _, value := range h.Values("x-codex-beta-features") { + if strings.TrimSpace(value) != "" { + return true + } + } + return false +} + +// applyOpenAICodexBetaFeatures 按真实 Codex 的会话级行为补注 +// x-codex-beta-features。 +// +// codex 侧规则（codex-rs：session/mod.rs build_model_client_beta_features_header +// 组装、client.rs build_responses_headers 附加）：该头是**会话级常量**，挂在 +// /responses、WS 握手、/responses/compact 三处的**每一个**请求上，而不是只挂 +// 压缩回合。其内容是"已启用且需要 advertise 的 feature 列表"，RemoteCompactionV2 +// 被特判 advertise；实测默认安装下没有任何 Experimental 特性默认开启， +// 因此默认 Codex 的头值恰好就是单个 "remote_compaction_v2"。 +// +// 网关据此对齐： +// - 原生 v2 压缩回合（body 带 compaction_trigger 实锤）：无论账号类型都确保 +// v2 在列，覆盖中间网关裁剪 token 的情形（#5586）； +// - ChatGPT codex 上游（OAuth）的其余请求：客户端**未**声明该头时补成默认 +// Codex 形态，消除"仅压缩回合才带该头"这种真实 Codex 不会产生的模式； +// - 客户端已声明该头：原样保留。非空但不含 v2 表示用户显式关闭了该特性， +// 网关不得替其改写能力声明； +// - 非 OAuth 上游（API Key/第三方兼容网关）：不做会话级注入，只保留压缩回合 +// 的那一条，避免向非 Codex 后端撒 Codex 专属头。 +// +// 已知无解的歧义：用户关掉 v2 且无其他特性时，真实 Codex 同样不发该头，与"老 +// 客户端"在线型上不可区分，此时按默认形态补注。该用户的 legacy 压缩端点本就 +// 已被上游下线（404），不存在可回退的正确行为。 +func applyOpenAICodexBetaFeatures(c *gin.Context, account *Account, h http.Header) { + if h == nil { + return + } + if isOpenAINativeCompactionV2(c) { + ensureOpenAIRemoteCompactionV2BetaFeature(h) + return + } + if account == nil || !account.IsOpenAIOAuth() { + return + } + if hasOpenAICodexBetaFeaturesHeader(h) { + return + } + h.Set("x-codex-beta-features", openAIRemoteCompactionV2Feature) +} // HasCompactionTriggerInInput detects an input item with // type="compaction_trigger". The handler combines this body signal with the diff --git a/backend/internal/service/openai_compact_probe.go b/backend/internal/service/openai_compact_probe.go index e8deff2d6d..9d5e8ccad4 100644 --- a/backend/internal/service/openai_compact_probe.go +++ b/backend/internal/service/openai_compact_probe.go @@ -10,7 +10,8 @@ import ( const ( // AccountTestModeDefault drives the standard /responses connection test. AccountTestModeDefault = "default" - // AccountTestModeCompact drives the /responses/compact compact-probe test. + // AccountTestModeCompact drives the remote-compaction probe test + // (native v2: streaming /responses with a compaction_trigger input item). AccountTestModeCompact = "compact" ) @@ -23,8 +24,12 @@ func normalizeAccountTestMode(mode string) string { } } -func createOpenAICompactProbePayload(model string) map[string]any { - return map[string]any{ +// createOpenAICompactProbePayload 构造原生 remote compaction v2 探测载荷： +// 流式 /responses + input 末尾 {"type":"compaction_trigger"}。上游已下线 +// legacy unary /responses/compact（v1 形态恒 404，#5598/#5624），现行 codex +// 默认协议即 v2（RemoteCompactionV2 Stable + default_enabled）。 +func createOpenAICompactProbePayload(model string, isOAuth bool) map[string]any { + payload := map[string]any{ "model": strings.TrimSpace(model), "instructions": "You are a helpful coding assistant.", "input": []any{ @@ -33,8 +38,35 @@ func createOpenAICompactProbePayload(model string) map[string]any { "role": "user", "content": "Respond with OK.", }, + map[string]any{"type": "compaction_trigger"}, }, + "stream": true, } + // ChatGPT internal API 要求 store: false，与真实转发一致。 + if isOAuth { + payload["store"] = false + } + return payload +} + +// openAICompactProbeFoundCompactionItem 判定探测响应是否产出了 compaction +// 输出 item——v2 契约的核心（codex 缺它即 fatal "got 0 items"）。三种形态都 +// 认：① SSE 的 output_item.done/added（原生 v2 主形态，codex 只从这里收集 +// item）；② SSE 终态 response.completed 的 response.output[]（部分上游只在 +// 终态给出 item）；③ 整体 JSON 的 output[]（老网关链把请求降级成 unary）。 +func openAICompactProbeFoundCompactionItem(body []byte) bool { + if len(body) == 0 { + return false + } + bodyText := string(body) + if _, found := findRawCompactionItemFromSSE(bodyText); found { + return true + } + if finalResponse, ok := extractCodexFinalResponse(bodyText); ok && + responsesOutputHasCompactionItem(finalResponse) { + return true + } + return responsesOutputHasCompactionItem(body) } func shouldMarkOpenAICompactUnsupported(status int, body []byte) bool { @@ -60,7 +92,12 @@ func shouldMarkOpenAICompactUnsupported(status int, body []byte) bool { return false } -func buildOpenAICompactProbeExtraUpdates(resp *http.Response, body []byte, probeErr error, now time.Time) map[string]any { +// buildOpenAICompactProbeExtraUpdates 计算探测结果的账号 extra 更新。 +// compactionFound 是 v2 契约判据：HTTP 2xx 但响应无 compaction item 时同样 +// 记为不支持（链路把 compaction_trigger 吞掉的形态，等价 codex 的 "got 0 +// items" fatal，#5478/#5648）。极端场景（上游链只支持 legacy unary compact） +// 可用账号级 openai_compact_mode=force_on 人工覆盖。 +func buildOpenAICompactProbeExtraUpdates(resp *http.Response, body []byte, probeErr error, compactionFound bool, now time.Time) map[string]any { updates := map[string]any{ "openai_compact_checked_at": now.Format(time.RFC3339), "openai_compact_last_status": nil, @@ -84,10 +121,14 @@ func buildOpenAICompactProbeExtraUpdates(resp *http.Response, body []byte, probe errMsg = "HTTP " + strconv.Itoa(resp.StatusCode) } errMsg = truncateString(sanitizeUpstreamErrorMessage(errMsg), 2048) - if resp.StatusCode >= 200 && resp.StatusCode < 300 { + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300 && compactionFound: updates["openai_compact_supported"] = true updates["openai_compact_last_error"] = "" - } else { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + updates["openai_compact_supported"] = false + updates["openai_compact_last_error"] = "upstream returned 2xx without a compaction output item (native remote compaction v2 unsupported)" + default: if shouldMarkOpenAICompactUnsupported(resp.StatusCode, body) { updates["openai_compact_supported"] = false } @@ -112,9 +153,14 @@ func mergeExtraUpdates(base map[string]any, more map[string]any) map[string]any return out } +// compactProbeSessionID 返回探测请求使用的会话标识。真实 Codex 的 +// session-id / thread-id 恒为 UUID（codex-protocol ThreadId 是 UUIDv7）， +// 探测既然与真实流量走同一个 /responses 端点，标识形态就必须同构—— +// 否则上游能凭 "probe_compact_5" 这类字面量一眼区分出探测流量。 +// 账号级稳定派生：重复探测复用同一会话，而不是每次新开一个。 func compactProbeSessionID(accountID int64) string { if accountID <= 0 { - return "probe_compact" + return deriveStableUUIDv4("sub2api:codex-compact-probe:v1:anonymous") } - return "probe_compact_" + strconv.FormatInt(accountID, 10) + return deriveStableUUIDv4("sub2api:codex-compact-probe:v1:" + strconv.FormatInt(accountID, 10)) } diff --git a/backend/internal/service/openai_compact_probe_test.go b/backend/internal/service/openai_compact_probe_test.go index fe3ba0e8d1..c273152e33 100644 --- a/backend/internal/service/openai_compact_probe_test.go +++ b/backend/internal/service/openai_compact_probe_test.go @@ -28,7 +28,7 @@ func TestNormalizeAccountTestMode(t *testing.T) { func TestBuildOpenAICompactProbeExtraUpdates_SuccessMarksSupported(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) - updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusOK}, []byte(`{"id":"cmp_1"}`), nil, now) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusOK}, []byte(`{"id":"cmp_1"}`), nil, true, now) if got := updates["openai_compact_supported"]; got != true { t.Fatalf("openai_compact_supported = %v, want true", got) @@ -47,7 +47,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_SuccessMarksSupported(t *testing.T) func TestBuildOpenAICompactProbeExtraUpdates_404MarksUnsupported(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) body := []byte(`404 page not found`) - updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusNotFound}, body, nil, now) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusNotFound}, body, nil, false, now) if got := updates["openai_compact_supported"]; got != false { t.Fatalf("openai_compact_supported = %v, want false", got) @@ -59,7 +59,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_404MarksUnsupported(t *testing.T) { func TestBuildOpenAICompactProbeExtraUpdates_502DoesNotMarkUnsupported(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) - updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusBadGateway}, []byte(`Upstream request failed`), nil, now) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusBadGateway}, []byte(`Upstream request failed`), nil, false, now) if _, exists := updates["openai_compact_supported"]; exists { t.Fatalf("did not expect openai_compact_supported for 502 response") @@ -71,7 +71,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_502DoesNotMarkUnsupported(t *testin func TestBuildOpenAICompactProbeExtraUpdates_RequestErrorDoesNotMarkUnsupported(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) - updates := buildOpenAICompactProbeExtraUpdates(nil, nil, errors.New("dial tcp timeout"), now) + updates := buildOpenAICompactProbeExtraUpdates(nil, nil, errors.New("dial tcp timeout"), false, now) if _, exists := updates["openai_compact_supported"]; exists { t.Fatalf("did not expect openai_compact_supported for request error") @@ -86,7 +86,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_RequestErrorDoesNotMarkUnsupported( func TestBuildOpenAICompactProbeExtraUpdates_NoResponseClearsLastStatus(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) - updates := buildOpenAICompactProbeExtraUpdates(nil, nil, nil, now) + updates := buildOpenAICompactProbeExtraUpdates(nil, nil, nil, false, now) if got, exists := updates["openai_compact_last_status"]; !exists || got != nil { t.Fatalf("openai_compact_last_status = %v, want nil key", got) @@ -99,7 +99,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_NoResponseClearsLastStatus(t *testi func TestBuildOpenAICompactProbeExtraUpdates_UnknownModelDoesNotMarkUnsupported(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) body := []byte(`{"error":{"message":"unknown model gpt-5.4-openai-compact"}}`) - updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusBadRequest}, body, nil, now) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusBadRequest}, body, nil, false, now) if _, exists := updates["openai_compact_supported"]; exists { t.Fatalf("did not expect openai_compact_supported for unknown-model diagnostics") @@ -111,7 +111,7 @@ func TestBuildOpenAICompactProbeExtraUpdates_UnknownModelDoesNotMarkUnsupported( func TestBuildOpenAICompactProbeExtraUpdates_EmptyFailureBodyFallsBackToHTTPStatus(t *testing.T) { now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) - updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusServiceUnavailable}, nil, nil, now) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusServiceUnavailable}, nil, nil, false, now) if got := updates["openai_compact_last_status"]; got != http.StatusServiceUnavailable { t.Fatalf("openai_compact_last_status = %v, want %d", got, http.StatusServiceUnavailable) @@ -120,3 +120,79 @@ func TestBuildOpenAICompactProbeExtraUpdates_EmptyFailureBodyFallsBackToHTTPStat t.Fatalf("openai_compact_last_error = %v, want HTTP 503", got) } } + +func TestBuildOpenAICompactProbeExtraUpdates_2xxWithoutCompactionItemMarksUnsupported(t *testing.T) { + now := time.Date(2026, 4, 10, 10, 0, 0, 0, time.UTC) + updates := buildOpenAICompactProbeExtraUpdates(&http.Response{StatusCode: http.StatusOK}, []byte(`{"id":"resp_1","output":[]}`), nil, false, now) + + if got := updates["openai_compact_supported"]; got != false { + t.Fatalf("openai_compact_supported = %v, want false（2xx 无 compaction item = v2 不可用）", got) + } + if got := updates["openai_compact_last_error"]; got == "" { + t.Fatalf("expected openai_compact_last_error to explain the missing compaction item") + } +} + +func TestCreateOpenAICompactProbePayload_NativeV2Shape(t *testing.T) { + payload := createOpenAICompactProbePayload("gpt-5.6-sol", true) + if payload["stream"] != true { + t.Fatalf("v2 probe payload must be streaming") + } + if payload["store"] != false { + t.Fatalf("OAuth probe payload must carry store:false") + } + input, ok := payload["input"].([]any) + if !ok || len(input) != 2 { + t.Fatalf("expected 2 input items, got %v", payload["input"]) + } + last, ok := input[len(input)-1].(map[string]any) + if !ok || last["type"] != "compaction_trigger" { + t.Fatalf("last input item must be compaction_trigger, got %v", input[len(input)-1]) + } + + apiKeyPayload := createOpenAICompactProbePayload("gpt-5.6-sol", false) + if _, has := apiKeyPayload["store"]; has { + t.Fatalf("API-key probe payload must not force store") + } +} + +func TestOpenAICompactProbeFoundCompactionItem(t *testing.T) { + sseWithItem := []byte("event: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"compaction\",\"id\":\"cmp_1\",\"encrypted_content\":\"blob\"}}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"output\":[]}}\n\n") + if !openAICompactProbeFoundCompactionItem(sseWithItem) { + t.Fatalf("SSE output_item.done 携带 compaction item 应判定为支持") + } + + sseAlias := []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"compaction_summary\",\"id\":\"cmp_2\"}}\n\n") + if !openAICompactProbeFoundCompactionItem(sseAlias) { + t.Fatalf("compaction_summary 别名应判定为支持") + } + + sseNoItem := []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_1\"}}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_2\",\"output\":[]}}\n\n") + if openAICompactProbeFoundCompactionItem(sseNoItem) { + t.Fatalf("无 compaction item 的流不应判定为支持") + } + + jsonWithItem := []byte(`{"id":"resp_3","output":[{"type":"compaction","id":"cmp_3"}]}`) + if !openAICompactProbeFoundCompactionItem(jsonWithItem) { + t.Fatalf("JSON output[] 携带 compaction item 应判定为支持（降级链兜底）") + } + + if openAICompactProbeFoundCompactionItem(nil) { + t.Fatalf("空响应不应判定为支持") + } +} + +func TestOpenAICompactProbeFoundCompactionItem_TerminalResponseOutput(t *testing.T) { + // 部分上游只在终态 response.completed 的 output[] 给出 compaction item， + // 事件流里没有 output_item.done——探针同样必须判定为支持。 + sseTerminalOnly := []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_1\"}}\n\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_t\",\"output\":[{\"type\":\"compaction\",\"id\":\"cmp_t\"}]}}\n\n") + if !openAICompactProbeFoundCompactionItem(sseTerminalOnly) { + t.Fatalf("终态 response.output 中的 compaction item 应判定为支持") + } + + sseTerminalEmpty := []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_e\",\"output\":[]}}\n\n") + if openAICompactProbeFoundCompactionItem(sseTerminalEmpty) { + t.Fatalf("终态 output 为空且无 item 事件时不应判定为支持") + } +} diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index c53d9e8241..4c02cb30f1 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -99,6 +99,12 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( } } } + // 真实 Codex 的 WS 握手同样携带会话级 x-codex-beta-features + // （client.rs build_websocket_headers 复用 build_responses_headers）， + // 客户端未声明时补成默认形态，与 HTTP 出站保持一致。放在客户端头拷贝 + // 之外：该头是账号/会话级属性，不依赖入站请求是否存在，也避免预热与 + // 实际请求因头差异落进不同的连接池兼容分桶。 + applyOpenAICodexBetaFeatures(c, account, headers) // OAuth 账号：将 apiKeyID 混入 session 标识符，防止跨用户会话碰撞。 if account != nil && account.Type == AccountTypeOAuth { apiKeyID := getAPIKeyIDFromContext(c)