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 0000000000..d5598c3d30 --- /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 0000000000..960bab01b8 --- /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 cab620b0d6..82cd89f695 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 cd9dc8dbb9..a9200b10b7 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 e7ddaca80e..617c07c066 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)