fix(openai): relay x-codex-turn-state and guard cross-account echo

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.

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
+	})
+}

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 后端不注入")
+}

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 {

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

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)