From 8219dcfc87ac270fe11414a98e326b06f5b4309f Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 15 Aug 2026 16:34:47 +0800 Subject: [PATCH 1/4] fix(openai): relay x-codex-turn-state and guard cross-account echo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex captures x-codex-turn-state from /responses SSE, /responses/compact JSON and the WS handshake (codex-api sse/responses.rs, endpoint/compact.rs), then echoes it back on later requests of the same turn. The HTTP path dropped it because the header is not in the generic response allowlist, while the WS path already relayed it — an inconsistency that broke the protocol chain. Relay it explicitly at every commit point instead of widening the global allowlist (which would leak it into Anthropic/Gemini responses): - streaming, non-streaming and SSE-to-JSON handlers relay it, clearing any value left over by a previous failover attempt when upstream sends none - under the first-output guard the header is only staged; provenance is recorded when applyAttemptResponseHeaders actually writes it, because a first-output timeout discards the staged headers and the client never receives that blob - record (api key + client session) -> minting account, TTL-bounded with an opportunistic sweep, and strip echoes known to come from another account before they go upstream. Stripping only: injection is the Claude bridge's job. Same-account or unknown provenance passes through unchanged. --- .../service/openai_codex_turn_state.go | 165 ++++++++ .../service/openai_codex_turn_state_test.go | 376 ++++++++++++++++++ .../openai_gateway_response_handling.go | 23 +- .../service/openai_gateway_service.go | 5 + .../service/openai_gateway_service_test.go | 8 +- 5 files changed, 569 insertions(+), 8 deletions(-) create mode 100644 backend/internal/service/openai_codex_turn_state.go create mode 100644 backend/internal/service/openai_codex_turn_state_test.go diff --git a/backend/internal/service/openai_codex_turn_state.go b/backend/internal/service/openai_codex_turn_state.go new file mode 100644 index 000000000..d5598c3d3 --- /dev/null +++ b/backend/internal/service/openai_codex_turn_state.go @@ -0,0 +1,165 @@ +package service + +import ( + "net/http" + "strconv" + "strings" + "time" + + "github.com/gin-gonic/gin" +) + +// openAICodexTurnStateHeader 是 Codex 的回合状态头。上游在响应头中铸造该 +// 不透明 blob，客户端在同一回合的后续请求中原样回带（codex-rs 侧从 +// /responses SSE、/responses/compact JSON 与 WS 握手三种响应中捕获，见 +// codex-api/src/sse/responses.rs 与 endpoint/compact.rs）。 +const openAICodexTurnStateHeader = "x-codex-turn-state" + +// turn-state blob 是上游在"出站身份"（含 #5553 指纹收敛改写后的 +// installation/session/thread 标识）下铸造的，同账号回放自洽；跨账号回放 +// （failover 换号后客户端仍回带旧账号的 blob）是代理链独有、真实 Codex +// 永远不会产生的矛盾信号。溯源表记录每个下游会话最近一次铸造该 blob 的 +// 账号，出站守卫据此剥离已知异账号的回带值。 +type openAICodexTurnStateOrigin struct { + accountID int64 + expiresAt time.Time +} + +// openAICodexTurnStateSeed 返回溯源表键：API Key + 客户端原始会话标识。 +// 客户端会话标识取自请求头（与指纹收敛的 thread 派生同源，见 +// extractClientSessionID），确保同一下游会话的记录/守卫两侧使用同一键。 +// 无会话标识时返回空串，表示不做跟踪（保持透传现状）。 +func openAICodexTurnStateSeed(c *gin.Context) string { + if c == nil || c.Request == nil { + return "" + } + sessionID := extractClientSessionID(c.Request.Header) + if sessionID == "" { + return "" + } + return strconv.FormatInt(getAPIKeyIDFromContext(c), 10) + "\x00" + sessionID +} + +// relayOpenAICodexTurnState 将上游响应中的 turn-state 显式写入下游响应头， +// 并记录铸造账号。必须在响应头提交点调用（WriteHeader 之前、且确认本次 +// 上游响应就是将要写回客户端的响应之后）。上游无该头时主动清除 writer 上 +// 可能残留的上一 failover attempt 的值——否则换号后旧账号的 blob 会粘到 +// 新账号的响应上，这正是本文件要防止的跨账号矛盾。 +func (s *OpenAIGatewayService) relayOpenAICodexTurnState(c *gin.Context, account *Account, upstream http.Header) { + if c == nil || c.Writer == nil { + return + } + canonical := http.CanonicalHeaderKey(openAICodexTurnStateHeader) + state := extractOpenAICodexTurnState(upstream) + if state == "" { + c.Writer.Header().Del(canonical) + return + } + c.Writer.Header().Set(canonical, state) + s.noteOpenAICodexTurnStateProvenance(c, account) +} + +// stageOpenAICodexTurnState 将上游 turn-state 暂存到延迟提交的响应头集合 +// （首输出守卫路径先缓存头、见到首个输出事件才提交）。此处**不**记录铸造 +// 账号：该 attempt 仍可能在首输出超时后 failover，暂存头会被整体丢弃， +// 客户端从未收到该 blob。溯源必须在真正提交时记录，见 +// noteStagedOpenAICodexTurnStateCommitted。 +func stageOpenAICodexTurnState(dst *http.Header, upstream http.Header) { + if dst == nil { + return + } + canonical := http.CanonicalHeaderKey(openAICodexTurnStateHeader) + state := extractOpenAICodexTurnState(upstream) + if state == "" { + if *dst != nil { + dst.Del(canonical) + } + return + } + if *dst == nil { + *dst = http.Header{} + } + dst.Set(canonical, state) +} + +// noteStagedOpenAICodexTurnStateCommitted 在暂存响应头真正写入下游时记录 +// 铸造账号——只有此刻客户端才确定收到了该 blob，溯源表才与客户端持有的 +// 值一致（否则被 failover 丢弃的 attempt 会污染溯源，导致后续误剥离）。 +func (s *OpenAIGatewayService) noteStagedOpenAICodexTurnStateCommitted(c *gin.Context, account *Account, staged http.Header) { + if staged == nil || strings.TrimSpace(staged.Get(openAICodexTurnStateHeader)) == "" { + return + } + s.noteOpenAICodexTurnStateProvenance(c, account) +} + +func extractOpenAICodexTurnState(upstream http.Header) string { + if upstream == nil { + return "" + } + return strings.TrimSpace(upstream.Get(openAICodexTurnStateHeader)) +} + +// noteOpenAICodexTurnStateProvenance 记录（下游会话 → 铸造账号）。 +func (s *OpenAIGatewayService) noteOpenAICodexTurnStateProvenance(c *gin.Context, account *Account) { + if s == nil || account == nil || account.ID <= 0 { + return + } + seed := openAICodexTurnStateSeed(c) + if seed == "" { + return + } + s.openaiCodexTurnStateOrigins.Store(seed, openAICodexTurnStateOrigin{ + accountID: account.ID, + expiresAt: time.Now().Add(s.openAIWSSessionStickyTTL()), + }) + s.sweepOpenAICodexTurnStateOrigins() +} + +// guardOpenAICodexTurnStateEcho 出站守卫：客户端回带的 turn-state 若已知由 +// 其他账号铸造则剥离，同账号或无溯源记录时保持原样。只剥离、不注入—— +// /responses 路径的客户端是真实 Codex，会按自身回合语义自行回带；服务端 +// 注入是 Claude 兼容桥（无法回带的客户端）的专属行为。 +func (s *OpenAIGatewayService) guardOpenAICodexTurnStateEcho(c *gin.Context, account *Account, h http.Header) { + if s == nil || h == nil || account == nil { + return + } + if strings.TrimSpace(h.Get(openAICodexTurnStateHeader)) == "" { + return + } + seed := openAICodexTurnStateSeed(c) + if seed == "" { + return + } + raw, ok := s.openaiCodexTurnStateOrigins.Load(seed) + if !ok { + return + } + origin, ok := raw.(openAICodexTurnStateOrigin) + if !ok { + s.openaiCodexTurnStateOrigins.Delete(seed) + return + } + if !origin.expiresAt.IsZero() && time.Now().After(origin.expiresAt) { + s.openaiCodexTurnStateOrigins.Delete(seed) + return + } + if origin.accountID != account.ID { + h.Del(openAICodexTurnStateHeader) + } +} + +// sweepOpenAICodexTurnStateOrigins 机会式清扫过期溯源记录：每 256 次写入 +// 全量遍历一轮，防止仅靠读侧惰性删除导致的慢泄漏（会话键无上界）。 +func (s *OpenAIGatewayService) sweepOpenAICodexTurnStateOrigins() { + if s.openaiCodexTurnStateWrites.Add(1)%256 != 0 { + return + } + now := time.Now() + s.openaiCodexTurnStateOrigins.Range(func(key, value any) bool { + origin, ok := value.(openAICodexTurnStateOrigin) + if !ok || (!origin.expiresAt.IsZero() && now.After(origin.expiresAt)) { + s.openaiCodexTurnStateOrigins.Delete(key) + } + return true + }) +} diff --git a/backend/internal/service/openai_codex_turn_state_test.go b/backend/internal/service/openai_codex_turn_state_test.go new file mode 100644 index 000000000..960bab01b --- /dev/null +++ b/backend/internal/service/openai_codex_turn_state_test.go @@ -0,0 +1,376 @@ +package service + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func newTurnStateTestContext(t *testing.T, apiKeyID int64, sessionID string) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + if sessionID != "" { + c.Request.Header.Set("session_id", sessionID) + } + if apiKeyID > 0 { + c.Set("api_key", &APIKey{ID: apiKeyID}) + } + return c, rec +} + +func TestOpenAICodexTurnStateSeed(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-1") + require.Equal(t, "7\x00sess-1", openAICodexTurnStateSeed(c)) + + // 连字符形式优先（Codex CLI 标准头） + c.Request.Header.Set("session-id", "sess-hyphen") + require.Equal(t, "7\x00sess-hyphen", openAICodexTurnStateSeed(c)) + + // 无会话标识 → 不跟踪 + cNoSession, _ := newTurnStateTestContext(t, 7, "") + require.Empty(t, openAICodexTurnStateSeed(cNoSession)) + + require.Empty(t, openAICodexTurnStateSeed(nil)) +} + +func TestRelayOpenAICodexTurnState_SetsHeaderAndRecordsProvenance(t *testing.T) { + svc := &OpenAIGatewayService{} + account := &Account{ID: 42} + c, _ := newTurnStateTestContext(t, 7, "sess-relay") + + upstream := http.Header{} + upstream.Set("x-codex-turn-state", "blob-A") + svc.relayOpenAICodexTurnState(c, account, upstream) + + require.Equal(t, "blob-A", c.Writer.Header().Get("X-Codex-Turn-State")) + + raw, ok := svc.openaiCodexTurnStateOrigins.Load("7\x00sess-relay") + require.True(t, ok) + origin := raw.(openAICodexTurnStateOrigin) + require.Equal(t, int64(42), origin.accountID) + require.True(t, origin.expiresAt.After(time.Now())) +} + +func TestRelayOpenAICodexTurnState_ClearsStaleValueWhenUpstreamAbsent(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-stale") + // 模拟上一 failover attempt 残留的值 + c.Writer.Header().Set("X-Codex-Turn-State", "blob-old") + + svc.relayOpenAICodexTurnState(c, &Account{ID: 43}, http.Header{}) + + require.Empty(t, c.Writer.Header().Get("X-Codex-Turn-State")) + _, ok := svc.openaiCodexTurnStateOrigins.Load("7\x00sess-stale") + require.False(t, ok) +} + +func TestStageOpenAICodexTurnState_StagedHeaders(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 9, "sess-staged") + + // nil 集合 + 上游有值 → 创建集合并写入，但此刻还不记录溯源 + var staged http.Header + upstream := http.Header{} + upstream.Set("x-codex-turn-state", "blob-B") + stageOpenAICodexTurnState(&staged, upstream) + require.NotNil(t, staged) + require.Equal(t, "blob-B", staged.Get("X-Codex-Turn-State")) + _, noted := svc.openaiCodexTurnStateOrigins.Load("9\x00sess-staged") + require.False(t, noted, "暂存阶段不得记录溯源：该 attempt 仍可能 failover 丢弃") + + // 真正提交时才记录 + svc.noteStagedOpenAICodexTurnStateCommitted(c, &Account{ID: 44}, staged) + raw, ok := svc.openaiCodexTurnStateOrigins.Load("9\x00sess-staged") + require.True(t, ok) + require.Equal(t, int64(44), raw.(openAICodexTurnStateOrigin).accountID) + + // 上游无值 → 清除已暂存的值；nil 集合保持 nil + stageOpenAICodexTurnState(&staged, http.Header{}) + require.Empty(t, staged.Get("X-Codex-Turn-State")) + var nilStaged http.Header + stageOpenAICodexTurnState(&nilStaged, http.Header{}) + require.Nil(t, nilStaged) +} + +// 首输出超时导致 attempt 被丢弃时，溯源不得被该 attempt 污染——否则后续 +// 请求会把客户端持有的合法 blob 误判成跨账号回带而剥离。 +func TestStagedTurnState_AbandonedAttemptDoesNotPoisonProvenance(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 11, "sess-abandoned") + + // 账号 A 的 attempt 暂存了 blob，但从未提交（首输出超时 → failover） + var staged http.Header + upstreamA := http.Header{} + upstreamA.Set("x-codex-turn-state", "blob-A") + stageOpenAICodexTurnState(&staged, upstreamA) + + // 账号 B 接手并真正提交 + svc.relayOpenAICodexTurnState(c, &Account{ID: 52}, upstreamA) + + // 客户端回带的 blob 来自 B，出站到 B 时不得被剥离 + h := http.Header{} + h.Set("x-codex-turn-state", "blob-A") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 52}, h) + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) + + raw, ok := svc.openaiCodexTurnStateOrigins.Load("11\x00sess-abandoned") + require.True(t, ok) + require.Equal(t, int64(52), raw.(openAICodexTurnStateOrigin).accountID) +} + +func TestNoteStagedOpenAICodexTurnStateCommitted_NoopWithoutState(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 12, "sess-nostate") + + svc.noteStagedOpenAICodexTurnStateCommitted(c, &Account{ID: 60}, nil) + svc.noteStagedOpenAICodexTurnStateCommitted(c, &Account{ID: 60}, http.Header{"X-Request-Id": []string{"rid"}}) + + _, ok := svc.openaiCodexTurnStateOrigins.Load("12\x00sess-nostate") + require.False(t, ok) +} + +func TestGuardOpenAICodexTurnStateEcho(t *testing.T) { + newOutbound := func(state string) http.Header { + h := http.Header{} + if state != "" { + h.Set("x-codex-turn-state", state) + } + return h + } + + t.Run("same_account_keeps_echo", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-g1") + upstream := http.Header{} + upstream.Set("x-codex-turn-state", "blob-A") + svc.relayOpenAICodexTurnState(c, &Account{ID: 42}, upstream) + + h := newOutbound("blob-A") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 42}, h) + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) + }) + + t.Run("foreign_account_strips_echo", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-g2") + upstream := http.Header{} + upstream.Set("x-codex-turn-state", "blob-A") + svc.relayOpenAICodexTurnState(c, &Account{ID: 42}, upstream) + + // failover 换到账号 43：blob 由 42 铸造，必须剥离 + h := newOutbound("blob-A") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) + require.Empty(t, h.Get("x-codex-turn-state")) + }) + + t.Run("no_provenance_passthrough", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-g3") + h := newOutbound("blob-unknown") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) + require.Equal(t, "blob-unknown", h.Get("x-codex-turn-state")) + }) + + t.Run("expired_provenance_passthrough_and_pruned", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-g4") + svc.openaiCodexTurnStateOrigins.Store("7\x00sess-g4", openAICodexTurnStateOrigin{ + accountID: 42, + expiresAt: time.Now().Add(-time.Minute), + }) + h := newOutbound("blob-A") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) + _, ok := svc.openaiCodexTurnStateOrigins.Load("7\x00sess-g4") + require.False(t, ok) + }) + + t.Run("no_session_seed_noop", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "") + h := newOutbound("blob-A") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) + }) + + t.Run("no_echo_noop", func(t *testing.T) { + svc := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "sess-g5") + h := newOutbound("") + svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) + require.Empty(t, h.Get("x-codex-turn-state")) + }) +} + +func TestSweepOpenAICodexTurnStateOrigins_PrunesExpiredEntries(t *testing.T) { + svc := &OpenAIGatewayService{} + svc.openaiCodexTurnStateOrigins.Store("expired", openAICodexTurnStateOrigin{ + accountID: 1, + expiresAt: time.Now().Add(-time.Minute), + }) + svc.openaiCodexTurnStateOrigins.Store("alive", openAICodexTurnStateOrigin{ + accountID: 2, + expiresAt: time.Now().Add(time.Hour), + }) + + // 计数器推进到触发清扫的边界 + svc.openaiCodexTurnStateWrites.Store(255) + svc.sweepOpenAICodexTurnStateOrigins() + + _, expiredOK := svc.openaiCodexTurnStateOrigins.Load("expired") + require.False(t, expiredOK) + _, aliveOK := svc.openaiCodexTurnStateOrigins.Load("alive") + require.True(t, aliveOK) +} + +func TestWriteOpenAIPassthroughResponseHeaders_RelaysAndClearsTurnState(t *testing.T) { + // filter=nil 走 content-type 兜底分支；turn-state 强制放行不依赖 filter。 + dst := http.Header{} + src := http.Header{} + src.Set("X-Codex-Turn-State", "blob-P") + writeOpenAIPassthroughResponseHeaders(dst, src, nil) + require.Equal(t, "blob-P", dst.Get("X-Codex-Turn-State")) + + // 上游缺失时清除残留（failover 换号防串扰） + writeOpenAIPassthroughResponseHeaders(dst, http.Header{"Content-Type": []string{"application/json"}}, nil) + require.Empty(t, dst.Get("X-Codex-Turn-State")) +} + +func TestEnsureOpenAIRemoteCompactionV2BetaFeature(t *testing.T) { + t.Run("absent_sets_feature", func(t *testing.T) { + h := http.Header{} + ensureOpenAIRemoteCompactionV2BetaFeature(h) + require.Equal(t, "remote_compaction_v2", h.Get("x-codex-beta-features")) + }) + + t.Run("present_unchanged", func(t *testing.T) { + h := http.Header{} + h.Set("x-codex-beta-features", "responses_websockets_v2, remote_compaction_v2") + ensureOpenAIRemoteCompactionV2BetaFeature(h) + require.Equal(t, "responses_websockets_v2, remote_compaction_v2", h.Get("x-codex-beta-features")) + }) + + t.Run("other_tokens_merged", func(t *testing.T) { + h := http.Header{} + h.Set("x-codex-beta-features", "responses_websockets_v2") + ensureOpenAIRemoteCompactionV2BetaFeature(h) + require.Equal(t, "responses_websockets_v2,remote_compaction_v2", h.Get("x-codex-beta-features")) + }) + + t.Run("multi_line_values_merged_single_line", func(t *testing.T) { + h := http.Header{} + h.Add("x-codex-beta-features", "feature_a") + h.Add("x-codex-beta-features", "feature_b") + ensureOpenAIRemoteCompactionV2BetaFeature(h) + require.Equal(t, []string{"feature_a,feature_b,remote_compaction_v2"}, h.Values("x-codex-beta-features")) + }) +} + +// 对齐真实 Codex：该头是会话级常量，挂在 OAuth 的每个请求上，而不是只在 +// 压缩回合出现（codex-rs build_model_client_beta_features_header）。 +func TestApplyOpenAICodexBetaFeatures(t *testing.T) { + oauthAccount := &Account{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + apiKeyAccount := &Account{ID: 2, Platform: PlatformOpenAI, Type: AccountTypeAPIKey} + + t.Run("oauth_plain_request_gets_default_codex_shape", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + h := http.Header{} + applyOpenAICodexBetaFeatures(c, oauthAccount, h) + require.Equal(t, "remote_compaction_v2", h.Get("x-codex-beta-features"), + "OAuth 的普通请求也必须带会话级 beta 头") + }) + + t.Run("client_declared_header_preserved", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + h := http.Header{} + h.Set("x-codex-beta-features", "some_other_feature") + applyOpenAICodexBetaFeatures(c, oauthAccount, h) + require.Equal(t, "some_other_feature", h.Get("x-codex-beta-features"), + "客户端显式声明的能力集不得被网关改写（非空即视为用户已关闭 v2）") + }) + + t.Run("native_v2_forces_feature_even_when_client_trimmed_it", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + MarkOpenAINativeCompactionV2(c) + h := http.Header{} + h.Set("x-codex-beta-features", "some_other_feature") + applyOpenAICodexBetaFeatures(c, oauthAccount, h) + require.Contains(t, h.Get("x-codex-beta-features"), "remote_compaction_v2", + "body 带 compaction_trigger 是实锤，必须确保 v2 在列") + require.Contains(t, h.Get("x-codex-beta-features"), "some_other_feature") + }) + + t.Run("native_v2_applies_to_non_oauth_too", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + MarkOpenAINativeCompactionV2(c) + h := http.Header{} + applyOpenAICodexBetaFeatures(c, apiKeyAccount, h) + require.Equal(t, "remote_compaction_v2", h.Get("x-codex-beta-features")) + }) + + t.Run("non_oauth_plain_request_untouched", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + h := http.Header{} + applyOpenAICodexBetaFeatures(c, apiKeyAccount, h) + require.Empty(t, h.Get("x-codex-beta-features"), + "非 Codex 后端不做会话级注入") + }) + + t.Run("nil_account_plain_request_untouched", func(t *testing.T) { + c, _ := newTurnStateTestContext(t, 7, "sess-beta") + h := http.Header{} + applyOpenAICodexBetaFeatures(c, nil, h) + require.Empty(t, h.Get("x-codex-beta-features")) + }) +} + +// WS 握手与 HTTP 出站必须给出同一份会话级 beta 头：真实 Codex 的 +// build_websocket_headers 复用 build_responses_headers（client.rs）， +// 两侧不一致还会让预热连接与实际请求落进不同的连接池兼容分桶。 +func TestBuildOpenAIWSHeaders_CarriesSessionBetaFeatures(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := &OpenAIGatewayService{} + decision := OpenAIWSProtocolDecision{Transport: OpenAIUpstreamTransportResponsesWebsocketV2} + + build := func(t *testing.T, account *Account, clientBeta string) http.Header { + t.Helper() + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + if clientBeta != "" { + c.Request.Header.Set("x-codex-beta-features", clientBeta) + } + headers, _, err := svc.buildOpenAIWSHeaders( + context.Background(), c, account, "test-token", decision, + true, "", "", "", "gpt-5.6-codex", "", + ) + require.NoError(t, err) + return headers + } + + oauthAccount := &Account{ + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{"chatgpt_account_id": "test-account"}, + } + + headers := build(t, oauthAccount, "") + require.Equal(t, "remote_compaction_v2", headers.Get("x-codex-beta-features"), + "WS 握手也必须带会话级 beta 头") + + declared := build(t, oauthAccount, "some_other_feature") + require.Equal(t, []string{"some_other_feature"}, declared.Values("x-codex-beta-features"), + "客户端已声明时原样保留") + + apiKeyHeaders := build(t, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}, "") + require.Empty(t, apiKeyHeaders.Get("x-codex-beta-features"), + "非 Codex 后端不注入") +} diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index cab620b0d..82cd89f69 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -64,6 +64,14 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } else if s.responseHeaderFilter != nil { responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) } + // x-codex-turn-state 不在通用响应头白名单内，按 Codex 协议显式回传： + // 客户端会在同回合的后续请求中回带（openai_codex_turn_state.go）。 + // 首输出守卫模式下只暂存，溯源在 applyAttemptResponseHeaders 真正提交时记录。 + if guardFirstOutput { + stageOpenAICodexTurnState(&attemptResponseHeaders, resp.Header) + } else { + s.relayOpenAICodexTurnState(c, account, resp.Header) + } // Set SSE response headers c.Header("Content-Type", "text/event-stream") @@ -85,6 +93,9 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. c.Writer.Header().Add(key, value) } } + // 暂存头此刻才真正写给客户端：turn-state 溯源在这里记录（见 + // noteStagedOpenAICodexTurnStateCommitted 的 failover 说明）。 + s.noteStagedOpenAICodexTurnStateCommitted(c, account, attemptResponseHeaders) // These headers describe this gateway's SSE stream and are stable across // account attempts. Keep them authoritative over upstream values. c.Header("Content-Type", "text/event-stream") @@ -1219,7 +1230,7 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r // Some OpenAI-compatible upstreams (including other sub2api instances) // may return SSE even when stream=false was requested. if isEventStreamResponse(resp.Header) { - return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel) + return s.handleSSEToJSON(resp, c, account, body, originalModel, mappedModel) } // bodyLooksLikeSSE is a line-level heuristic: real SSE framing requires // "data:"/"event:" field names at the very start of a physical line. A @@ -1235,7 +1246,7 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r // positives on JSON responses that coincidentally contain "data:" or // "event:" in their text content. if account.Type == AccountTypeOAuth && bodyLooksLikeSSE { - return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel) + return s.handleSSEToJSON(resp, c, account, body, originalModel, mappedModel) } if account != nil && account.IsGrok() && isOpenAIResponsesCompactPath(c) { body, err = convertGrokResponseToOpenAICompact(body) @@ -1247,7 +1258,7 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r usageValue, usageOK := extractOpenAIUsageFromJSONBytes(body) if !usageOK { if bodyLooksLikeSSE { - return s.handleSSEToJSON(resp, c, body, originalModel, mappedModel) + return s.handleSSEToJSON(resp, c, account, body, originalModel, mappedModel) } return nil, fmt.Errorf("parse response: invalid json response") } @@ -1266,6 +1277,9 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r return nil, fmt.Errorf("restore OpenAI namespace response: %w", err) } responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + // Codex 协议要求 /responses/compact JSON 响应携带 x-codex-turn-state + // （codex-api/src/endpoint/compact.rs 从响应头捕获），显式回传。 + s.relayOpenAICodexTurnState(c, account, resp.Header) contentType := "application/json" if s.cfg != nil && !s.cfg.Security.ResponseHeaders.Enabled { @@ -1309,7 +1323,7 @@ func bodyHasSSEFraming(body []byte) bool { return false } -func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Context, body []byte, originalModel, mappedModel string) (*openaiNonStreamingResult, error) { +func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Context, account *Account, body []byte, originalModel, mappedModel string) (*openaiNonStreamingResult, error) { bodyText := string(body) finalResponse, ok := extractCodexFinalResponse(bodyText) @@ -1361,6 +1375,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte } responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) + s.relayOpenAICodexTurnState(c, account, resp.Header) contentType := "application/json; charset=utf-8" if !ok { diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index cd9dc8dbb..a9200b10b 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -462,6 +462,11 @@ type OpenAIGatewayService struct { codexModelsManifestCache codexModelsManifestCache openaiCompatSessionResponses sync.Map openaiCompatAnthropicDigestSessions sync.Map + // openaiCodexTurnStateOrigins: 下游会话 seed → openAICodexTurnStateOrigin， + // 记录最近一次向该会话下发 x-codex-turn-state 的铸造账号，供出站守卫 + // 剥离跨账号回带（openai_codex_turn_state.go）。 + openaiCodexTurnStateOrigins sync.Map + openaiCodexTurnStateWrites atomic.Uint64 } // NewOpenAIGatewayService creates a new OpenAIGatewayService diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index e7ddaca80..617c07c06 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -3475,7 +3475,7 @@ func TestHandleSSEToJSON_CompletedEventReturnsJSON(t *testing.T) { `data: [DONE]`, }, "\n")) - usage, err := svc.handleSSEToJSON(resp, c, body, "gpt-4o", "gpt-4o") + usage, err := svc.handleSSEToJSON(resp, c, nil, body, "gpt-4o", "gpt-4o") require.NoError(t, err) require.NotNil(t, usage) require.Equal(t, 7, usage.InputTokens) @@ -3594,7 +3594,7 @@ func TestHandleSSEToJSON_ReconstructsImageGenerationOutputItemDone(t *testing.T) `data: [DONE]`, }, "\n")) - usage, err := svc.handleSSEToJSON(resp, c, body, "gpt-5.4", "gpt-5.4") + usage, err := svc.handleSSEToJSON(resp, c, nil, body, "gpt-5.4", "gpt-5.4") require.NoError(t, err) require.NotNil(t, usage) require.Equal(t, 4, usage.ImageOutputTokens) @@ -3621,7 +3621,7 @@ func TestHandleSSEToJSON_NoFinalResponseKeepsSSEBody(t *testing.T) { `data: [DONE]`, }, "\n")) - usage, err := svc.handleSSEToJSON(resp, c, body, "gpt-4o", "gpt-4o") + usage, err := svc.handleSSEToJSON(resp, c, nil, body, "gpt-4o", "gpt-4o") require.NoError(t, err) require.NotNil(t, usage) require.Equal(t, 0, usage.InputTokens) @@ -3645,7 +3645,7 @@ func TestHandleSSEToJSON_ResponseFailedReturnsProtocolError(t *testing.T) { `data: [DONE]`, }, "\n")) - usage, err := svc.handleSSEToJSON(resp, c, body, "gpt-4o", "gpt-4o") + usage, err := svc.handleSSEToJSON(resp, c, nil, body, "gpt-4o", "gpt-4o") require.Nil(t, usage) require.Error(t, err) require.Equal(t, http.StatusBadGateway, rec.Code) From 8ae6d8f67e72b099ed581b1455840ca62bb25561 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 15 Aug 2026 16:35:08 +0800 Subject: [PATCH 2/4] 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 19373b4c4..1695eccc1 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 9e1d29c2f..496ef57ae 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 5b973285a..5b4772508 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 5dad35887..a8f234f58 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 5e7957c55..83266e6c0 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 e8deff2d6..9d5e8ccad 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 fe3ba0e8d..c273152e3 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 c53d9e824..4c02cb30f 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) From fce41e318f147cca42d04944754ed4e693714ddf Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 15 Aug 2026 16:35:26 +0800 Subject: [PATCH 3/4] fix(openai): make Codex fingerprint convergence opt-in and cover passthrough Default codex_fingerprint_mode to off. v0.1.175 treated a missing key as "session", so upgrading silently rewrote installation/session/thread/turn/ window identifiers for every existing OAuth account that had never configured this field. The quota regressions in #5555, #5556 and #5582 line up with that version boundary, with A/B reports that rolling back to v0.1.173 restores quota. Convergence is now explicit opt-in (#5610). Only accounts that never set the field change behaviour; explicit off / device / session / full keep working exactly as configured. That required flipping the persistence condition in all three account modals from "!== 'session'" to "!== 'off'": the old rule deleted the key when it equalled the default, which after the flip would have silently discarded an administrator's explicit opt-in to session. Also extend convergence to the passthrough path, which previously left client identifiers untouched: - resolve the ids once in forwardOpenAIPassthrough and rewrite client_metadata on the raw bytes (gjson extract + sjson splice) because passthrough is a hot path that must not fully unmarshal multi-MB bodies; a shared core keeps the raw and map variants from drifting - both request builders apply the staged ids at the same relative position (after session isolation, before identity enforcement) so headers and body share one id set and turn_id stays consistent - stage the ids unconditionally, including nil: a failover from a converged account to an off account must not leave the previous account's ids behind --- .../service/openai_codex_fingerprint.go | 108 ++++++++- .../service/openai_codex_fingerprint_test.go | 219 +++++++++++++++++- .../service/openai_gateway_forward.go | 22 +- .../service/openai_gateway_passthrough.go | 49 ++++ .../account/BulkEditAccountModal.vue | 7 +- .../components/account/CreateAccountModal.vue | 8 +- .../components/account/EditAccountModal.vue | 12 +- .../src/i18n/locales/en/admin/accounts.ts | 6 +- .../src/i18n/locales/zh/admin/accounts.ts | 6 +- 9 files changed, 392 insertions(+), 45 deletions(-) diff --git a/backend/internal/service/openai_codex_fingerprint.go b/backend/internal/service/openai_codex_fingerprint.go index 55564c534..c74669c00 100644 --- a/backend/internal/service/openai_codex_fingerprint.go +++ b/backend/internal/service/openai_codex_fingerprint.go @@ -9,9 +9,43 @@ import ( "strings" "time" + "github.com/gin-gonic/gin" "github.com/google/uuid" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" ) +// codexFingerprintIDsContextKey 是暂存在 gin context 的收敛 ID 集合键。 +// 由 Forward（非透传）或 forwardOpenAIPassthrough（透传）解析后写入，请求 +// 构造器读取用于出站头改写——请求体与出站头必须共享同一份 IDs，保证 +// turn_id 等随机字段一致。 +const codexFingerprintIDsContextKey = "codex_fingerprint_ids" + +// stageCodexFingerprintIDs 将本 attempt 解析出的收敛 ID 暂存到 gin context。 +// 必须无条件覆写（含 nil）：failover 从收敛账号切到 off 账号时，上一账号的 +// IDs 不得残留并被误应用到新账号的出站头（typed-nil 由应用侧 nil 守卫吸收）。 +func stageCodexFingerprintIDs(c *gin.Context, ids *codexFingerprintIDs) { + if c != nil { + c.Set(codexFingerprintIDsContextKey, ids) + } +} + +// applyStagedCodexFingerprintHeaders 读取 context 暂存的收敛 ID 并改写出站头。 +// 非透传与透传两个请求构造器共用本函数，防止应用语义漂移。仅 OAuth 账号 +// 生效（stale 键在账号类型混合 failover 下由该门挡住）。 +func applyStagedCodexFingerprintHeaders(c *gin.Context, account *Account, h http.Header) { + if c == nil || account == nil || account.Type != AccountTypeOAuth { + return + } + value, ok := c.Get(codexFingerprintIDsContextKey) + if !ok { + return + } + if ids, ok := value.(*codexFingerprintIDs); ok { + applyCodexFingerprintHeaders(h, ids) + } +} + // codexFingerprintMode 控制 OAuth 账号出站请求的设备指纹收敛强度。 // 多人共享同一 OAuth 账号时，每个用户的 Codex 客户端会携带各自不同的 // installation_id / session_id / thread_id，上游据此判定设备数和会话数。 @@ -19,7 +53,8 @@ import ( type codexFingerprintMode string const ( - // codexFingerprintOff 不做任何收敛，原样透传客户端标识（默认行为）。 + // codexFingerprintOff 不做任何收敛，原样透传客户端标识。 + // 这是默认值：收敛是显式 opt-in 的（见 GetCodexFingerprintMode）。 codexFingerprintOff codexFingerprintMode = "off" // codexFingerprintDevice 仅收敛 installation_id 为账号级恒定值。 // 上游看到 1 台设备 + 多会话（每用户各自的 session）。 @@ -36,7 +71,16 @@ const ( const codexFingerprintModeExtraKey = "codex_fingerprint_mode" // GetCodexFingerprintMode 从账号 extra JSON 读取指纹收敛模式。 -// 未设置时默认 session（设备+会话收敛），显式设为 "off" 才关闭。 +// +// **收敛是显式 opt-in**：未设置、空值或非法值一律按 off 处理，只有管理员 +// 明确配置 device / session / full 才收敛。 +// +// 历史：v0.1.175（#5553）把缺省值当作 session，导致升级后存量 OAuth 账号 +// （普遍没有这个 extra 键）的每个非透传请求都被静默改写 installation / +// session / thread / turn / window 五类标识；#5555、#5556、#5582 报告的额度 +// 缩水都卡在该版本边界，并有"回退 v0.1.173 即恢复"与"新账号开收敛后降额" +// 的 A/B 实测。上游的配额判定策略不可观测，因此这里取兼容安全的一侧： +// 不显式 opt-in 就保持 v0.1.175 之前的客户端身份（#5610）。 func (a *Account) GetCodexFingerprintMode() codexFingerprintMode { if a == nil || !a.IsOpenAIOAuth() { return codexFingerprintOff @@ -46,7 +90,7 @@ func (a *Account) GetCodexFingerprintMode() codexFingerprintMode { case codexFingerprintOff, codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull: return codexFingerprintMode(raw) default: - return codexFingerprintSession + return codexFingerprintOff } } @@ -245,6 +289,21 @@ func applyCodexFingerprintClientMetadata(reqBody map[string]any, ids *codexFinge existing = make(map[string]any) } + if !applyCodexFingerprintToClientMetadataMap(existing, ids) { + return false + } + reqBody["client_metadata"] = existing + return true +} + +// applyCodexFingerprintToClientMetadataMap 是 client_metadata 改写的共享核心， +// map 版（非透传，body 已解码）与 raw 字节版（透传热路径）都经由它，保证两条 +// 路径的收敛语义永不漂移。 +func applyCodexFingerprintToClientMetadataMap(existing map[string]any, ids *codexFingerprintIDs) bool { + if existing == nil || ids == nil { + return false + } + modified := false if ids.installationID != "" { @@ -256,9 +315,6 @@ func applyCodexFingerprintClientMetadata(reqBody map[string]any, ids *codexFinge rewriteClientMetadataEmbeddedTurnMetadata(existing, map[string]any{ "installation_id": ids.installationID, }) - if modified { - reqBody["client_metadata"] = existing - } return modified } @@ -276,11 +332,47 @@ func applyCodexFingerprintClientMetadata(reqBody map[string]any, ids *codexFinge "window_id": ids.windowID, "turn_started_at_unix_ms": time.Now().UnixMilli(), }) - - reqBody["client_metadata"] = existing return true } +// applyCodexFingerprintClientMetadataRaw 在原始 JSON 字节上改写 client_metadata， +// 供透传路径使用——透传是热路径，禁止对可能高达数十 MB 的 body 做全量 +// Unmarshal（见 forwardOpenAIPassthrough 的轻量提取注释）。实现为：gjson 提取 +// client_metadata 小对象单独解码，经共享核心改写后 sjson 一次性拼回，body +// 其余字节原样保留。语义与 applyCodexFingerprintClientMetadata 逐点一致 +// （含"非对象值整体替换为收敛集合"的行为）。 +func applyCodexFingerprintClientMetadataRaw(body []byte, ids *codexFingerprintIDs) ([]byte, bool, error) { + if len(body) == 0 || ids == nil { + return body, false, nil + } + // 非 JSON 对象的 body（数组/标量/畸形）没有 client_metadata 语义， + // sjson 在这类根上写字段会改写整体结构，直接放行保持原样。 + if !gjson.ParseBytes(body).IsObject() { + return body, false, nil + } + + existing := map[string]any{} + if cm := gjson.GetBytes(body, "client_metadata"); cm.IsObject() { + if err := json.Unmarshal([]byte(cm.Raw), &existing); err != nil { + return body, false, fmt.Errorf("decode client_metadata for fingerprint: %w", err) + } + } + + if !applyCodexFingerprintToClientMetadataMap(existing, ids) { + return body, false, nil + } + + raw, err := json.Marshal(existing) + if err != nil { + return body, false, fmt.Errorf("encode converged client_metadata: %w", err) + } + next, err := sjson.SetRawBytes(body, "client_metadata", raw) + if err != nil { + return body, false, fmt.Errorf("splice converged client_metadata: %w", err) + } + return next, true, nil +} + // rewriteClientMetadataEmbeddedTurnMetadata 改写 client_metadata 中内嵌的 // x-codex-turn-metadata JSON 字符串里的指定字段。 func rewriteClientMetadataEmbeddedTurnMetadata(clientMetadata map[string]any, fields map[string]any) { diff --git a/backend/internal/service/openai_codex_fingerprint_test.go b/backend/internal/service/openai_codex_fingerprint_test.go index 039f843c2..8e74bfa8b 100644 --- a/backend/internal/service/openai_codex_fingerprint_test.go +++ b/backend/internal/service/openai_codex_fingerprint_test.go @@ -1,10 +1,13 @@ package service import ( + "context" "encoding/json" "net/http" + "net/http/httptest" "testing" + "github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -51,9 +54,11 @@ func TestGetCodexFingerprintMode(t *testing.T) { }{ {"nil 账号", nil, codexFingerprintOff}, {"非 OAuth 账号", &Account{Platform: PlatformOpenAI, Type: "api_key"}, codexFingerprintOff}, - {"无 extra 默认 session", newTestOAuthAccount(1, nil), codexFingerprintSession}, - {"空值默认 session", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: ""}), codexFingerprintSession}, - {"非法值默认 session", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: "invalid"}), codexFingerprintSession}, + // 收敛是显式 opt-in：缺省/空/非法一律 off（#5610）。存量账号普遍没有这个 + // extra 键，升级不得把它们静默切进收敛。 + {"无 extra 默认 off", newTestOAuthAccount(1, nil), codexFingerprintOff}, + {"空值默认 off", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: ""}), codexFingerprintOff}, + {"非法值默认 off", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: "invalid"}), codexFingerprintOff}, {"显式 off", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: "off"}), codexFingerprintOff}, {"device", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: "device"}), codexFingerprintDevice}, {"session", newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: "session"}), codexFingerprintSession}, @@ -116,13 +121,24 @@ func TestResolveCodexFingerprintIDsFromRequest_ExplicitOff(t *testing.T) { assert.Nil(t, ids, "显式 off 模式应返回 nil") } -func TestResolveCodexFingerprintIDsFromRequest_DefaultIsSession(t *testing.T) { +// 未显式配置的存量账号不得被收敛（#5610）：默认返回 nil，出站身份保持 +// v0.1.175 之前的客户端原值。 +func TestResolveCodexFingerprintIDsFromRequest_DefaultIsOff(t *testing.T) { account := newTestOAuthAccount(1, nil) - ids := resolveCodexFingerprintIDsFromRequest(account, nil) - require.NotNil(t, ids, "无 extra 默认 session 模式，应返回非 nil") - assert.Equal(t, codexFingerprintSession, ids.mode) - assert.NotEmpty(t, ids.sessionID) - assert.NotEmpty(t, ids.turnID) + assert.Nil(t, resolveCodexFingerprintIDsFromRequest(account, nil), "无 extra 应视为 off") +} + +// 管理员显式 opt-in 的账号行为不变。 +func TestResolveCodexFingerprintIDsFromRequest_ExplicitOptInHonored(t *testing.T) { + for _, mode := range []string{"device", "session", "full"} { + t.Run(mode, func(t *testing.T) { + account := newTestOAuthAccount(1, map[string]any{codexFingerprintModeExtraKey: mode}) + ids := resolveCodexFingerprintIDsFromRequest(account, nil) + require.NotNil(t, ids, "显式配置必须生效") + assert.Equal(t, codexFingerprintMode(mode), ids.mode) + assert.NotEmpty(t, ids.installationID) + }) + } } // --- applyCodexFingerprintHeaders: off 模式 --- @@ -458,3 +474,188 @@ func TestExtractClientSessionID(t *testing.T) { }) } } + +// --- 透传路径：raw 字节版 client_metadata 改写 --- + +// rawVsMapClientMetadata 用同一份 ids 分别跑 map 版与 raw 字节版， +// 返回两侧最终的 client_metadata 解码结果。 +func rawVsMapClientMetadata(t *testing.T, body []byte, ids *codexFingerprintIDs) (map[string]any, map[string]any) { + t.Helper() + + var decoded map[string]any + require.NoError(t, json.Unmarshal(body, &decoded)) + applyCodexFingerprintClientMetadata(decoded, ids) + mapCM, _ := decoded["client_metadata"].(map[string]any) + + rawBody, changed, err := applyCodexFingerprintClientMetadataRaw(body, ids) + require.NoError(t, err) + require.True(t, changed) + var rawDecoded map[string]any + require.NoError(t, json.Unmarshal(rawBody, &rawDecoded)) + rawCM, _ := rawDecoded["client_metadata"].(map[string]any) + return mapCM, rawCM +} + +func TestApplyCodexFingerprintClientMetadataRaw_MatchesMapVariant(t *testing.T) { + embedded := `{\"installation_id\":\"real-install\",\"session_id\":\"real-session\",\"sandbox\":\"seatbelt\"}` + bodies := map[string]string{ + "no_client_metadata": `{"model":"gpt-5.6-sol","input":[],"stream":true}`, + "object_with_extras": `{"model":"gpt-5.6-sol","client_metadata":{"session_id":"client-session","traceparent":"00-abc-def-01","x-codex-turn-metadata":"` + embedded + `"},"stream":true}`, + "non_object_value": `{"model":"gpt-5.6-sol","client_metadata":"bogus","stream":true}`, + } + for _, mode := range []codexFingerprintMode{codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull} { + account := newTestOAuthAccount(4242, nil) + ids := resolveCodexFingerprintIDs(account, "client-sess-raw", mode) + require.NotNil(t, ids) + for name, body := range bodies { + t.Run(string(mode)+"/"+name, func(t *testing.T) { + mapCM, rawCM := rawVsMapClientMetadata(t, []byte(body), ids) + assert.Equal(t, mapCM, rawCM, "raw 字节版与 map 版的 client_metadata 结果必须逐点一致") + }) + } + } +} + +func TestApplyCodexFingerprintClientMetadataRaw_PreservesUnrelatedFields(t *testing.T) { + account := newTestOAuthAccount(4243, nil) + ids := resolveCodexFingerprintIDs(account, "client-sess-preserve", codexFingerprintSession) + require.NotNil(t, ids) + + body := []byte(`{"model":"gpt-5.6-sol","input":[{"type":"message","role":"user","content":"hi"}],"stream":true,"prompt_cache_key":"pck-1"}`) + out, changed, err := applyCodexFingerprintClientMetadataRaw(body, ids) + require.NoError(t, err) + require.True(t, changed) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(out, &decoded)) + assert.Equal(t, "gpt-5.6-sol", decoded["model"]) + assert.Equal(t, "pck-1", decoded["prompt_cache_key"]) + assert.Equal(t, true, decoded["stream"]) + cm, _ := decoded["client_metadata"].(map[string]any) + require.NotNil(t, cm) + assert.Equal(t, ids.sessionID, cm["session_id"]) + assert.Equal(t, ids.turnID, cm["turn_id"]) +} + +func TestApplyCodexFingerprintClientMetadataRaw_Noop(t *testing.T) { + body := []byte(`{"model":"gpt-5.6-sol"}`) + out, changed, err := applyCodexFingerprintClientMetadataRaw(body, nil) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, body, out) + + out, changed, err = applyCodexFingerprintClientMetadataRaw(nil, &codexFingerprintIDs{mode: codexFingerprintSession, installationID: "x"}) + require.NoError(t, err) + assert.False(t, changed) + assert.Nil(t, out) +} + +// --- context 暂存与出站头应用（透传/非透传共用 seam）--- + +func newFingerprintStageTestContext(t *testing.T) *gin.Context { + t.Helper() + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + return c +} + +func TestStageCodexFingerprintIDs_NilOverwritesPreviousAccount(t *testing.T) { + c := newFingerprintStageTestContext(t) + accountA := newTestOAuthAccount(1001, nil) + idsA := resolveCodexFingerprintIDs(accountA, "sess-x", codexFingerprintSession) + require.NotNil(t, idsA) + stageCodexFingerprintIDs(c, idsA) + + // failover 切到 off 模式账号：无条件覆写为 nil，上一账号 IDs 不得残留 + stageCodexFingerprintIDs(c, nil) + + h := http.Header{} + h.Set("session_id", "isolated-session") + accountB := newTestOAuthAccount(1002, map[string]any{"codex_fingerprint_mode": "off"}) + applyStagedCodexFingerprintHeaders(c, accountB, h) + assert.Equal(t, "isolated-session", h.Get("session_id"), "off 账号不得应用上一账号的收敛 ID") + assert.Empty(t, h.Get("x-codex-installation-id")) +} + +func TestApplyStagedCodexFingerprintHeaders_SkipsNonOAuthAccount(t *testing.T) { + c := newFingerprintStageTestContext(t) + oauthIDs := resolveCodexFingerprintIDs(newTestOAuthAccount(1003, nil), "sess-y", codexFingerprintSession) + require.NotNil(t, oauthIDs) + stageCodexFingerprintIDs(c, oauthIDs) + + h := http.Header{} + apiKeyAccount := &Account{ID: 1004, Platform: PlatformOpenAI, Type: AccountTypeAPIKey} + applyStagedCodexFingerprintHeaders(c, apiKeyAccount, h) + assert.Empty(t, h.Get("x-codex-installation-id"), "stale 收敛 ID 不得应用到非 OAuth 账号") +} + +func TestBuildUpstreamRequestOpenAIPassthrough_AppliesStagedFingerprint(t *testing.T) { + svc := &OpenAIGatewayService{} + // 收敛是显式 opt-in（#5610）：显式开启后验证透传路径的出站头收敛。 + account := newTestOAuthAccount(2001, map[string]any{ + "openai_oauth_passthrough": true, + "codex_fingerprint_mode": "session", + }) + + c := newFingerprintStageTestContext(t) + c.Request.Header.Set("session_id", "real-client-session") + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1 (Ubuntu 22.4.0; x86_64) xterm-256color") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("x-codex-turn-metadata", `{"installation_id":"real-install","session_id":"real-session","sandbox":"seatbelt"}`) + + // 复刻 forwardOpenAIPassthrough 的解析+暂存 seam（默认 session 模式） + ids := resolveCodexFingerprintIDsFromRequest(account, c.Request.Header) + require.NotNil(t, ids) + stageCodexFingerprintIDs(c, ids) + + body := []byte(`{"model":"gpt-5.6-sol","input":[],"stream":true}`) + req, err := svc.buildUpstreamRequestOpenAIPassthrough(context.Background(), c, account, body, "test-token") + require.NoError(t, err) + + assert.Equal(t, ids.sessionID, req.Header.Get("session_id"), "session 模式下出站 session_id 应为账号级收敛值") + assert.Equal(t, ids.installationID, req.Header.Get("x-codex-installation-id")) + assert.Equal(t, ids.windowID, req.Header.Get("x-codex-window-id")) + assert.Equal(t, ids.threadID, req.Header.Get("x-client-request-id")) + turnMetadata := req.Header.Get("x-codex-turn-metadata") + require.NotEmpty(t, turnMetadata) + assert.Contains(t, turnMetadata, ids.sessionID, "turn-metadata JSON 中的 session_id 应被收敛") + assert.Contains(t, turnMetadata, `"sandbox":"seatbelt"`, "turn-metadata 未指定字段应原样保留") +} + +func TestBuildUpstreamRequestOpenAIPassthrough_OffModeKeepsIsolatedSession(t *testing.T) { + svc := &OpenAIGatewayService{} + account := newTestOAuthAccount(2002, map[string]any{ + "openai_oauth_passthrough": true, + "codex_fingerprint_mode": "off", + }) + + c := newFingerprintStageTestContext(t) + c.Request.Header.Set("session_id", "real-client-session") + c.Request.Header.Set("originator", "codex_cli_rs") + + ids := resolveCodexFingerprintIDsFromRequest(account, c.Request.Header) + require.Nil(t, ids) + stageCodexFingerprintIDs(c, ids) + + body := []byte(`{"model":"gpt-5.6-sol","input":[],"stream":true}`) + req, err := svc.buildUpstreamRequestOpenAIPassthrough(context.Background(), c, account, body, "test-token") + require.NoError(t, err) + + assert.NotEmpty(t, req.Header.Get("session_id")) + assert.NotEqual(t, resolveConvergedSessionID(account), req.Header.Get("session_id"), "off 模式不得收敛 session_id") + assert.Empty(t, req.Header.Get("x-codex-window-id")) +} + +func TestApplyCodexFingerprintClientMetadataRaw_NonObjectBodyUntouched(t *testing.T) { + account := newTestOAuthAccount(4244, nil) + ids := resolveCodexFingerprintIDs(account, "client-sess-nonobj", codexFingerprintSession) + require.NotNil(t, ids) + + for _, body := range []string{`[1,2,3]`, `"plain string"`, `not json at all`} { + out, changed, err := applyCodexFingerprintClientMetadataRaw([]byte(body), ids) + require.NoError(t, err) + assert.False(t, changed, "非 JSON 对象 body 不应被改写: %s", body) + assert.Equal(t, []byte(body), out) + } +} diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index aa54242eb..a3ba6ebab 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -427,10 +427,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco markDecodedModified() } } - // 将 fpIDs 存入 gin context，供 buildUpstreamRequest 中头改写使用 - if c != nil && fpIDs != nil { - c.Set("codex_fingerprint_ids", fpIDs) - } + // 将 fpIDs 存入 gin context，供 buildUpstreamRequest 中头改写使用。 + // 无条件覆写（含 nil）：failover 从收敛账号切到 off 账号时，上一 + // 账号的 IDs 不得残留（stageCodexFingerprintIDs 注释）。 + stageCodexFingerprintIDs(c, fpIDs) } if codexResult.NormalizedModel != "" { upstreamModel = codexResult.NormalizedModel @@ -1094,6 +1094,9 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. } } } + // 客户端回带的 x-codex-turn-state 若已知由其他账号铸造（failover 换号）， + // 剥离后再出站——异账号 blob 与本账号的（指纹收敛后）出站身份自相矛盾。 + s.guardOpenAICodexTurnStateEcho(c, account, req.Header) if account.Type == AccountTypeOAuth { compatMessagesBridge := isOpenAICompatMessagesBridgeContext(c) || isOpenAICompatMessagesBridgeBody(body) // 清除客户端透传的 session 头，后续用隔离后的值重新设置，防止跨用户会话碰撞。 @@ -1144,13 +1147,7 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. } // 指纹收敛：使用 Forward() 中预计算的收敛 ID 改写出站头，与请求体使用同一份 IDs。 - if account.Type == AccountTypeOAuth && c != nil { - if fpIDs, ok := c.Get("codex_fingerprint_ids"); ok { - if ids, ok := fpIDs.(*codexFingerprintIDs); ok { - applyCodexFingerprintHeaders(req.Header, ids) - } - } - } + applyStagedCodexFingerprintHeaders(c, account, req.Header) // 终态收口：强制统一 OAuth 出站身份（User-Agent / originator / version 同源自洽）。 // 客户端自报身份不参与构造，浏览器型 UA 也因此不会再到达上游（原浏览器 UA 兜底已被吸收）。 @@ -1165,6 +1162,9 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. // 账号级请求头覆写（仅 openai api_key 账号启用时生效；OAuth 路径 no-op） account.ApplyHeaderOverrides(req.Header) + // x-codex-beta-features：按真实 Codex 的会话级行为补注（在账号级覆写之后， + // 保证不被覆盖丢失）。 + applyOpenAICodexBetaFeatures(c, account, req.Header) setOpenAICodexRoutingHintFromBody(req.Header, account, body) logOpenAIRoutingDiagnosticsFromBody(ctx, account, "http", req.Header, body, "not_applicable") diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 8c029f0aa..1fac0440f 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -80,6 +80,28 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( body = normalizedBody } reqStream = gjson.GetBytes(body, "stream").Bool() + + // 指纹收敛：与非透传路径同门控（仅 OAuth、legacy compact 形态跳过）。 + // 一次性解析收敛 ID：请求体 client_metadata 在此改写（raw 字节外科 + // 手术，透传热路径禁全量 Unmarshal），出站头改写由请求构造器读取 + // context 中的同一份 IDs 完成（turn_id 等随机字段两侧必须一致）。 + if !isOpenAIResponsesCompactPath(c) { + var clientHeaders http.Header + if c != nil && c.Request != nil { + clientHeaders = c.Request.Header + } + fpIDs := resolveCodexFingerprintIDsFromRequest(account, clientHeaders) + if fpIDs != nil { + fpBody, fpChanged, fpErr := applyCodexFingerprintClientMetadataRaw(body, fpIDs) + if fpErr != nil { + return nil, fpErr + } + if fpChanged { + body = fpBody + } + } + stageCodexFingerprintIDs(c, fpIDs) + } } sanitizedBody, sanitized, err := sanitizeEmptyBase64InputImagesInOpenAIBody(body) @@ -239,6 +261,13 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( serviceTier := extractOpenAIServiceTierFromBody(body) + // x-codex-turn-state 溯源：下游回传由 writeOpenAIPassthroughResponseHeaders + // 在各 handler 的写头点强制放行，铸造账号在此统一记录，供出站守卫剥离 + // failover 换号后的跨账号回带（openai_codex_turn_state.go）。 + if extractOpenAICodexTurnState(resp.Header) != "" { + s.noteOpenAICodexTurnStateProvenance(c, account) + } + var usage *OpenAIUsage var firstTokenMs *int responseID := "" @@ -376,6 +405,10 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( } } + // 客户端回带的 x-codex-turn-state 若已知由其他账号铸造（failover 换号）， + // 剥离后再出站（openai_codex_turn_state.go）。 + s.guardOpenAICodexTurnStateEcho(c, account, req.Header) + // 覆盖入站鉴权残留，并注入上游认证 req.Header.Del("authorization") req.Header.Del("x-api-key") @@ -447,6 +480,10 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( if s.cfg != nil && s.cfg.Gateway.ForceCodexCLI { req.Header.Set("user-agent", codexCLIUserAgent) } + // 指纹收敛：使用 forwardOpenAIPassthrough 中预计算的收敛 ID 改写出站头， + // 与请求体 client_metadata 共享同一份 IDs（与非透传路径相同的相对位置： + // 会话隔离之后、终态身份收口之前）。 + applyStagedCodexFingerprintHeaders(c, account, req.Header) // 终态收口：透传路径的 OAuth 与非透传完全一致，同样强制统一出站身份 // （User-Agent / originator / version 同源自洽），客户端自报身份不会到达上游。 if account.Type == AccountTypeOAuth { @@ -459,6 +496,9 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( // 账号级请求头覆写（仅 openai api_key 账号启用时生效；OAuth 路径 no-op） account.ApplyHeaderOverrides(req.Header) + // x-codex-beta-features：按真实 Codex 的会话级行为补注（在账号级覆写之后， + // 保证不被覆盖丢失）。 + applyOpenAICodexBetaFeatures(c, account, req.Header) setOpenAICodexRoutingHintFromBody(req.Header, account, body) logOpenAIRoutingDiagnosticsFromBody(ctx, account, "http_passthrough", req.Header, body, "not_applicable") @@ -1650,4 +1690,13 @@ func writeOpenAIPassthroughResponseHeaders(dst http.Header, src http.Header, fil dst.Add(key, v) } } + + // x-codex-turn-state：Codex 回合状态头，客户端会在同回合后续请求回带。 + // 与上面的用量头不同，这里在上游缺失时也主动清除——failover 换号后残留 + // 上一账号的 blob 会构成跨账号矛盾（openai_codex_turn_state.go）。 + turnStateKey := http.CanonicalHeaderKey(openAICodexTurnStateHeader) + dst.Del(turnStateKey) + for _, v := range getCaseInsensitiveValues(src, openAICodexTurnStateHeader) { + dst.Add(turnStateKey, v) + } } diff --git a/frontend/src/components/account/BulkEditAccountModal.vue b/frontend/src/components/account/BulkEditAccountModal.vue index e7d251792..86379b6bb 100644 --- a/frontend/src/components/account/BulkEditAccountModal.vue +++ b/frontend/src/components/account/BulkEditAccountModal.vue @@ -1535,7 +1535,7 @@ const codexCLIOnlyEnabled = ref(false) const codexCLIOnlyAppServerEnabled = ref(false) type CodexFingerprintMode = 'off' | 'device' | 'session' | 'full' const enableCodexFingerprintMode = ref(false) -const codexFingerprintMode = ref('session') +const codexFingerprintMode = ref('off') const codexFingerprintModeOptions = computed(() => [ { value: 'off' as CodexFingerprintMode, label: t('admin.accounts.openai.codexFingerprintOff') }, { value: 'device' as CodexFingerprintMode, label: t('admin.accounts.openai.codexFingerprintDevice') }, @@ -1829,7 +1829,8 @@ const buildUpdatePayload = (): Record | null => { if (enableCodexFingerprintMode.value) { const extra = ensureExtra() - if (codexFingerprintMode.value !== 'session') { + // off = 默认值，清键即可；device/session/full 是显式 opt-in，必须落键（#5610）。 + if (codexFingerprintMode.value !== 'off') { extra.codex_fingerprint_mode = codexFingerprintMode.value } else { delete extra.codex_fingerprint_mode @@ -2082,7 +2083,7 @@ watch( enableCodexCLIOnly.value = false enableCodexCLIOnlyAppServer.value = false enableCodexFingerprintMode.value = false - codexFingerprintMode.value = 'session' + codexFingerprintMode.value = 'off' enableOpenAICompactMode.value = false enableOpenAICompactModelMapping.value = false enableRpmLimit.value = false diff --git a/frontend/src/components/account/CreateAccountModal.vue b/frontend/src/components/account/CreateAccountModal.vue index f231df8d5..f93b71e57 100644 --- a/frontend/src/components/account/CreateAccountModal.vue +++ b/frontend/src/components/account/CreateAccountModal.vue @@ -3855,7 +3855,7 @@ const openaiAPIKeyResponsesWebSocketV2Mode = ref(OPENAI_WS_MODE_OF const codexCLIOnlyEnabled = ref(false) const codexCLIOnlyAppServerEnabled = ref(false) type CodexFingerprintMode = 'off' | 'device' | 'session' | 'full' -const codexFingerprintMode = ref('session') +const codexFingerprintMode = ref('off') const codexFingerprintModeOptions = computed(() => [ { value: 'off' as CodexFingerprintMode, label: t('admin.accounts.openai.codexFingerprintOff') }, { value: 'device' as CodexFingerprintMode, label: t('admin.accounts.openai.codexFingerprintDevice') }, @@ -4741,7 +4741,7 @@ const resetForm = () => { openaiAPIKeyResponsesWebSocketV2Mode.value = OPENAI_WS_MODE_OFF codexCLIOnlyEnabled.value = false codexCLIOnlyAppServerEnabled.value = false - codexFingerprintMode.value = 'session' + codexFingerprintMode.value = 'off' anthropicPassthroughEnabled.value = false anthropicAPIKeyAuthScheme.value = 'x_api_key' webSearchEmulationMode.value = 'default' @@ -4840,7 +4840,9 @@ const buildOpenAIExtra = (base?: Record): Record(OPENAI_WS_MODE_OF const codexCLIOnlyEnabled = ref(false) const codexCLIOnlyAppServerEnabled = ref(false) type CodexFingerprintMode = 'off' | 'device' | 'session' | 'full' -const codexFingerprintMode = ref('session') +const codexFingerprintMode = ref('off') type CodexImageToolMode = 'inherit' | 'enabled' | 'disabled' | 'block' const codexImageToolMode = ref('inherit') type AnthropicAPIKeyAuthScheme = 'x_api_key' | 'authorization_bearer' @@ -3440,7 +3440,7 @@ const syncFormFromAccount = (newAccount: Account | null) => { openaiAPIKeyResponsesWebSocketV2Mode.value = OPENAI_WS_MODE_OFF codexCLIOnlyEnabled.value = false codexCLIOnlyAppServerEnabled.value = false - codexFingerprintMode.value = 'session' + codexFingerprintMode.value = 'off' codexImageToolMode.value = 'inherit' anthropicPassthroughEnabled.value = false anthropicAPIKeyAuthScheme.value = 'x_api_key' @@ -3494,9 +3494,10 @@ const syncFormFromAccount = (newAccount: Account | null) => { } if (newAccount.type === 'oauth') { const fpMode = extra?.codex_fingerprint_mode as string | undefined + // 缺省/非法值按 off 呈现，与后端 GetCodexFingerprintMode 的 opt-in 语义一致（#5610） codexFingerprintMode.value = (['off', 'device', 'session', 'full'].includes(fpMode || '') ? fpMode as CodexFingerprintMode - : 'session') + : 'off') } const credentials = newAccount.credentials as Record | undefined const compactMappings = credentials?.compact_model_mapping as Record | undefined @@ -4843,9 +4844,10 @@ const handleSubmit = async () => { } } - // 指纹收敛模式：默认 session，不写入；非默认值显式写入（包括 off） + // 指纹收敛模式：默认 off（不写入）；device/session/full 是显式 opt-in， + // 必须落键，否则管理员的选择会被后端当作"未设置"而回落到 off（#5610）。 if (props.account.type === 'oauth') { - if (codexFingerprintMode.value !== 'session') { + if (codexFingerprintMode.value !== 'off') { newExtra.codex_fingerprint_mode = codexFingerprintMode.value } else { delete newExtra.codex_fingerprint_mode diff --git a/frontend/src/i18n/locales/en/admin/accounts.ts b/frontend/src/i18n/locales/en/admin/accounts.ts index 5de4a2fcc..0eb245033 100644 --- a/frontend/src/i18n/locales/en/admin/accounts.ts +++ b/frontend/src/i18n/locales/en/admin/accounts.ts @@ -574,10 +574,10 @@ export default { codexCLIOnlyAppServerDesc: "Effective only when the switch above is on. When enabled, this account also allows third-party clients that embed the Codex engine over the app-server protocol (e.g. Claude Code's codex plugin); they still pass the global engine-fingerprint gate. OR-combined with the global app-server toggle.", codexFingerprintMode: 'Codex fingerprint convergence', - codexFingerprintModeDesc: 'When multiple users share the same OAuth account, converge device/session identifiers to account-level stable values to reduce upstream-visible device and session count. Off = pass through client identifiers as-is.', - codexFingerprintOff: 'Off (passthrough)', + codexFingerprintModeDesc: 'When multiple users share the same OAuth account, converge device/session identifiers to account-level stable values to reduce upstream-visible device and session count. Off by default (client identifiers pass through as-is); opt in explicitly when needed. Some accounts reported quota shrinkage after enabling convergence, so choose based on your own measurements.', + codexFingerprintOff: 'Off (passthrough, default)', codexFingerprintDevice: 'Device only', - codexFingerprintSession: 'Device + Session (recommended)', + codexFingerprintSession: 'Device + Session', codexFingerprintFull: 'Full convergence', codexImageTool: 'Codex image bridge policy', codexImageToolDesc: diff --git a/frontend/src/i18n/locales/zh/admin/accounts.ts b/frontend/src/i18n/locales/zh/admin/accounts.ts index 1359fb0ee..1c6d000c3 100644 --- a/frontend/src/i18n/locales/zh/admin/accounts.ts +++ b/frontend/src/i18n/locales/zh/admin/accounts.ts @@ -644,10 +644,10 @@ export default { codexCLIOnlyAppServer: '允许 Codex app-server 客户端', codexCLIOnlyAppServerDesc: '仅在上方开关开启时生效。开启后本账号额外放行内嵌 Codex 引擎、经 app-server 协议接入的第三方客户端（如 Claude Code 的 codex 插件），仍需通过全局引擎指纹门；与全局 app-server 开关取 OR（任一开即放行）。', codexFingerprintMode: 'Codex 指纹收敛', - codexFingerprintModeDesc: '多人共享同一 OAuth 账号时，将各用户的设备/会话标识收敛为账号级恒定值，减少上游可见的设备数和会话数。关闭时原样透传客户端标识。', - codexFingerprintOff: '关闭（透传）', + codexFingerprintModeDesc: '多人共享同一 OAuth 账号时，将各用户的设备/会话标识收敛为账号级恒定值，减少上游可见的设备数和会话数。默认关闭（原样透传客户端标识），需要时再显式开启；部分账号开启收敛后出现过额度缩水，请按自己的实测结果选择。', + codexFingerprintOff: '关闭（透传，默认）', codexFingerprintDevice: '仅设备', - codexFingerprintSession: '设备+会话（推荐）', + codexFingerprintSession: '设备+会话', codexFingerprintFull: '完全收敛', codexImageTool: 'Codex 图片桥接策略', codexImageToolDesc: From 4d9fedee204012ee51089259ce28e25add21e541 Mon Sep 17 00:00:00 2001 From: shaw Date: Sat, 15 Aug 2026 16:44:30 +0800 Subject: [PATCH 4/4] fix(test): check type assertions in turn-state provenance tests errcheck runs with check-type-assertions enabled, so the single-value assertions on the provenance map values fail lint. --- .../internal/service/openai_codex_turn_state_test.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/backend/internal/service/openai_codex_turn_state_test.go b/backend/internal/service/openai_codex_turn_state_test.go index 960bab01b..a58e16629 100644 --- a/backend/internal/service/openai_codex_turn_state_test.go +++ b/backend/internal/service/openai_codex_turn_state_test.go @@ -54,7 +54,8 @@ func TestRelayOpenAICodexTurnState_SetsHeaderAndRecordsProvenance(t *testing.T) raw, ok := svc.openaiCodexTurnStateOrigins.Load("7\x00sess-relay") require.True(t, ok) - origin := raw.(openAICodexTurnStateOrigin) + origin, ok := raw.(openAICodexTurnStateOrigin) + require.True(t, ok) require.Equal(t, int64(42), origin.accountID) require.True(t, origin.expiresAt.After(time.Now())) } @@ -90,7 +91,9 @@ func TestStageOpenAICodexTurnState_StagedHeaders(t *testing.T) { svc.noteStagedOpenAICodexTurnStateCommitted(c, &Account{ID: 44}, staged) raw, ok := svc.openaiCodexTurnStateOrigins.Load("9\x00sess-staged") require.True(t, ok) - require.Equal(t, int64(44), raw.(openAICodexTurnStateOrigin).accountID) + origin, ok := raw.(openAICodexTurnStateOrigin) + require.True(t, ok) + require.Equal(t, int64(44), origin.accountID) // 上游无值 → 清除已暂存的值；nil 集合保持 nil stageOpenAICodexTurnState(&staged, http.Header{}) @@ -123,7 +126,9 @@ func TestStagedTurnState_AbandonedAttemptDoesNotPoisonProvenance(t *testing.T) { raw, ok := svc.openaiCodexTurnStateOrigins.Load("11\x00sess-abandoned") require.True(t, ok) - require.Equal(t, int64(52), raw.(openAICodexTurnStateOrigin).accountID) + origin, ok := raw.(openAICodexTurnStateOrigin) + require.True(t, ok) + require.Equal(t, int64(52), origin.accountID) } func TestNoteStagedOpenAICodexTurnStateCommitted_NoopWithoutState(t *testing.T) {