diff --git a/backend/internal/service/openai_agent_identity_compat_test.go b/backend/internal/service/openai_agent_identity_compat_test.go index a8f234f58c..2520eaaaeb 100644 --- a/backend/internal/service/openai_agent_identity_compat_test.go +++ b/backend/internal/service/openai_agent_identity_compat_test.go @@ -133,8 +133,8 @@ func TestOpenAIAgentIdentityPassthroughKeepsSessionAndPromptCacheHeaders(t *test require.Equal(t, "account-agent-passthrough", req.Header.Get("chatgpt-account-id")) require.NotEqual(t, "client-session", req.Header.Get("session_id")) require.NotEqual(t, "client-conversation", req.Header.Get("conversation_id")) - require.Equal(t, isolateOpenAISessionID(0, "client-session"), req.Header.Get("session_id")) - require.Equal(t, isolateOpenAISessionID(0, "client-conversation"), req.Header.Get("conversation_id")) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "client-session"), req.Header.Get("session_id")) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "client-conversation"), req.Header.Get("conversation_id")) requestBody, err := io.ReadAll(req.Body) require.NoError(t, err) require.Contains(t, string(requestBody), `"prompt_cache_key":"cache-agent"`) @@ -147,7 +147,7 @@ func TestOpenAIAgentIdentityPassthroughKeepsSessionAndPromptCacheHeaders(t *test Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{ - "chatgpt_account_id": "account-oauth-passthrough", + "chatgpt_account_id": "account-agent-passthrough", }, } oauthRecorder := httptest.NewRecorder() diff --git a/backend/internal/service/openai_alpha_search.go b/backend/internal/service/openai_alpha_search.go index e4fa52fe56..ac242f050c 100644 --- a/backend/internal/service/openai_alpha_search.go +++ b/backend/internal/service/openai_alpha_search.go @@ -30,6 +30,9 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co if s == nil || c == nil || account == nil { return nil, fmt.Errorf("service, context, and account are required") } + if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { + return nil, err + } modelResult := gjson.GetBytes(body, "model") requestedModel := strings.TrimSpace(modelResult.String()) if modelResult.Type != gjson.String || requestedModel == "" { @@ -272,10 +275,11 @@ func (s *OpenAIGatewayService) buildOpenAIAlphaSearchResponsesWebSearchRequest(c } apiKeyID := getAPIKeyIDFromContext(c) if sessionID := strings.TrimSpace(gjson.GetBytes(alphaBody, "id").String()); sessionID != "" { - isolated := isolateOpenAISessionID(apiKeyID, sessionID) + isolated := isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), sessionID) req.Header.Set("Session_ID", isolated) req.Header.Set("Conversation_ID", isolated) } + applyCodexAccountIdentityHeaders(req.Header, codexAccountIdentitySource(c, account), apiKeyID) enforceCodexIdentityHeadersWithUA(req.Header, s.codexIdentityOverrideUA(account)) account.ApplyHeaderOverrides(req.Header) return req, nil @@ -393,6 +397,7 @@ func (s *OpenAIGatewayService) buildOpenAIAlphaSearchRequest(ctx context.Context if turnMetadata := openAIAlphaSearchInboundHeader(c, "X-Codex-Turn-Metadata"); turnMetadata != "" { req.Header.Set("X-Codex-Turn-Metadata", turnMetadata) } + applyCodexAccountIdentityHeaders(req.Header, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) canonical := resolveCodexOutboundIdentity("") if version := openAIAlphaSearchInboundHeader(c, "Version"); version != "" { req.Header.Set("Version", version) diff --git a/backend/internal/service/openai_alpha_search_test.go b/backend/internal/service/openai_alpha_search_test.go index b5d37d1dd8..8521c2160e 100644 --- a/backend/internal/service/openai_alpha_search_test.go +++ b/backend/internal/service/openai_alpha_search_test.go @@ -59,6 +59,7 @@ func TestForwardAlphaSearchOAuthPreservesWire(t *testing.T) { c.Request.Header.Set("User-Agent", codexCLIUserAgent) c.Request.Header.Set("Originator", "codex_cli_rs") c.Request.Header.Set("Version", "0.144.1") + c.Request.Header.Set("X-Codex-Turn-Metadata", `{"session_id":"search-session","turn_id":"search-turn"}`) upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, @@ -92,6 +93,14 @@ func TestForwardAlphaSearchOAuthPreservesWire(t *testing.T) { require.Equal(t, "application/json", upstream.lastReq.Header.Get("Accept")) require.Equal(t, codexCLIVersion, upstream.lastReq.Header.Get("Version")) require.Empty(t, upstream.lastReq.Header.Get("OpenAI-Beta")) + require.Equal(t, + scopeCodexAccountIdentityValue(account, 0, "session", "search-session"), + gjson.Get(upstream.lastReq.Header.Get("X-Codex-Turn-Metadata"), "session_id").String(), + ) + require.Equal(t, + scopeCodexAccountIdentityValue(account, 0, "turn", "search-turn"), + gjson.Get(upstream.lastReq.Header.Get("X-Codex-Turn-Metadata"), "turn_id").String(), + ) require.JSONEq(t, string(body), string(upstream.lastBody)) } @@ -156,7 +165,10 @@ func TestForwardAlphaSearchPATUsesResponsesWebSearchFallback(t *testing.T) { require.Equal(t, "text/event-stream", upstream.lastReq.Header.Get("Accept")) require.Equal(t, "responses=experimental", upstream.lastReq.Header.Get("OpenAI-Beta")) require.Equal(t, codexCLIVersion, upstream.lastReq.Header.Get("Version")) - require.Equal(t, `{"turn_id":"turn-1"}`, upstream.lastReq.Header.Get("X-Codex-Turn-Metadata")) + require.Equal(t, + scopeCodexAccountIdentityValue(account, 0, "turn", "turn-1"), + gjson.Get(upstream.lastReq.Header.Get("X-Codex-Turn-Metadata"), "turn_id").String(), + ) require.Equal(t, openai.CodexDefaultOriginator, upstream.lastReq.Header.Get("Originator")) require.Empty(t, upstream.lastReq.Header.Get("X-Codex-Beta-Features")) require.Empty(t, upstream.lastReq.Header.Get("X-Codex-Turn-State")) diff --git a/backend/internal/service/openai_codex_account_identity.go b/backend/internal/service/openai_codex_account_identity.go new file mode 100644 index 0000000000..d0e336ec6c --- /dev/null +++ b/backend/internal/service/openai_codex_account_identity.go @@ -0,0 +1,275 @@ +package service + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +const codexAccountIdentityNamespaceVersion = "v1" + +const codexAccountIdentitySourceContextKey = "openai_codex_account_identity_source" + +// prepareCodexAccountIdentitySource resolves credential shadows once per selected +// attempt. The handler reuses gin.Context across failover attempts, so every entry +// point overwrites the staged source before projecting outbound identity. +func (s *OpenAIGatewayService) prepareCodexAccountIdentitySource(ctx context.Context, c *gin.Context, account *Account) (*Account, error) { + source := account + if account != nil && account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return nil, err + } + source = resolved + } + if c != nil { + c.Set(codexAccountIdentitySourceContextKey, source) + } + return source, nil +} + +func codexAccountIdentitySource(c *gin.Context, fallback *Account) *Account { + if c != nil { + if staged, ok := c.Get(codexAccountIdentitySourceContextKey); ok { + if source, ok := staged.(*Account); ok && source != nil { + return source + } + } + } + return fallback +} + +// codexAccountIdentityNamespace returns a stable, credential-scoped namespace. +// Multiple local rows that use the same ChatGPT account intentionally share the +// same namespace. Setup tokens use an irreversible bearer fingerprint because +// they have no refresh lifecycle or imported account metadata. Refreshable OAuth +// otherwise falls back only to a persistent fingerprint seed: local row IDs are +// deployment-relative and must never become upstream identity. +func codexAccountIdentityNamespace(account *Account) string { + if account == nil || !account.IsOpenAIOAuthLike() { + return "" + } + if upstreamAccountID := strings.TrimSpace(account.GetChatGPTAccountID()); upstreamAccountID != "" { + if upstreamUserID := strings.TrimSpace(account.GetCredential("chatgpt_user_id")); upstreamUserID != "" { + return "chatgpt:" + upstreamAccountID + ":user:" + upstreamUserID + } + return "chatgpt:" + upstreamAccountID + } + if seed, ok := codexFingerprintSeed(account.Extra); ok { + return "seed:" + seed + } + if account.Type == AccountTypeSetupToken { + if token := strings.TrimSpace(account.GetOpenAIAccessToken()); token != "" { + sum := sha256.Sum256([]byte("openai-setup-token:" + token)) + return fmt.Sprintf("setup-token:%x", sum[:16]) + } + } + return "" +} + +// isolateOpenAIUpstreamSessionID preserves the existing API-key isolation while +// adding the selected OAuth credential namespace. A scheduler failover therefore +// cannot send the same session/conversation identity through two upstream accounts. +func isolateOpenAIUpstreamSessionID(apiKeyID int64, account *Account, raw string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + namespace := codexAccountIdentityNamespace(account) + if namespace == "" { + return isolateOpenAISessionID(apiKeyID, raw) + } + sum := sha256.Sum256([]byte(fmt.Sprintf("u%d:a%s:%s", apiKeyID, namespace, raw))) + return fmt.Sprintf("%x", sum[:8]) +} + +func scopeCodexAccountIdentityValue(account *Account, apiKeyID int64, kind, raw string) string { + raw = strings.TrimSpace(raw) + namespace := codexAccountIdentityNamespace(account) + if raw == "" || namespace == "" { + return raw + } + return deriveStableUUIDv4(fmt.Sprintf( + "sub2api:codex-account-identity:%s:user:%d:account:%s:kind:%s:value:%s", + codexAccountIdentityNamespaceVersion, + apiKeyID, + namespace, + kind, + raw, + )) +} + +var codexAccountIdentityFields = []struct { + name string + kind string +}{ + {name: "installation_id", kind: "installation"}, + {name: "x-codex-installation-id", kind: "installation"}, + {name: "session_id", kind: "session"}, + {name: "session-id", kind: "session"}, + {name: "thread_id", kind: "thread"}, + {name: "thread-id", kind: "thread"}, + {name: "turn_id", kind: "turn"}, + {name: "turn-id", kind: "turn"}, + {name: "window_id", kind: "window"}, + {name: "x-codex-window-id", kind: "window"}, + {name: "x-client-request-id", kind: "request"}, +} + +func applyCodexAccountIdentityFields(values map[string]any, account *Account, apiKeyID int64) bool { + if values == nil || codexAccountIdentityNamespace(account) == "" { + return false + } + changed := false + for _, field := range codexAccountIdentityFields { + raw, ok := values[field.name].(string) + if !ok || strings.TrimSpace(raw) == "" { + continue + } + next := scopeCodexAccountIdentityValue(account, apiKeyID, field.kind, raw) + if next != raw { + values[field.name] = next + changed = true + } + } + return changed +} + +func applyCodexAccountIdentityEmbeddedMetadata(values map[string]any, account *Account, apiKeyID int64) bool { + raw, ok := values[openAIWSTurnMetadataHeader].(string) + if !ok || strings.TrimSpace(raw) == "" { + return false + } + metadata := map[string]any{} + if err := json.Unmarshal([]byte(raw), &metadata); err != nil || metadata == nil { + return false + } + if !applyCodexAccountIdentityFields(metadata, account, apiKeyID) { + return false + } + rebuilt, err := json.Marshal(metadata) + if err != nil { + return false + } + values[openAIWSTurnMetadataHeader] = string(rebuilt) + return true +} + +func applyCodexAccountIdentityClientMetadataMap(requestBody map[string]any, account *Account, apiKeyID int64) bool { + if requestBody == nil || codexAccountIdentityNamespace(account) == "" { + return false + } + changed := false + clientMetadata, _ := requestBody["client_metadata"].(map[string]any) + originalBodySessionID := "" + if clientMetadata != nil { + originalBodySessionID, _ = clientMetadata["session_id"].(string) + if applyCodexAccountIdentityFields(clientMetadata, account, apiKeyID) { + changed = true + } + if applyCodexAccountIdentityEmbeddedMetadata(clientMetadata, account, apiKeyID) { + changed = true + } + } + if raw, ok := requestBody["prompt_cache_key"].(string); ok && strings.TrimSpace(raw) != "" { + kind := "prompt-cache" + if strings.TrimSpace(originalBodySessionID) != "" && raw == originalBodySessionID { + kind = "session" + } + next := scopeCodexAccountIdentityValue(account, apiKeyID, kind, raw) + if next != raw { + requestBody["prompt_cache_key"] = next + changed = true + } + } + return changed +} + +// applyCodexAccountIdentityClientMetadataRaw scopes only the small identity +// subobjects with gjson/sjson. The passthrough hot path never unmarshals the +// potentially multi-megabyte request body. +func applyCodexAccountIdentityClientMetadataRaw(body []byte, account *Account, apiKeyID int64) ([]byte, bool, error) { + if len(body) == 0 || codexAccountIdentityNamespace(account) == "" { + return body, false, nil + } + root := gjson.ParseBytes(body) + if !root.IsObject() { + return body, false, nil + } + + next := body + changed := false + originalBodySessionID := "" + if cm := gjson.GetBytes(body, "client_metadata"); cm.IsObject() { + clientMetadata := map[string]any{} + if err := json.Unmarshal([]byte(cm.Raw), &clientMetadata); err != nil { + return body, false, fmt.Errorf("decode client_metadata for account identity: %w", err) + } + originalBodySessionID, _ = clientMetadata["session_id"].(string) + metadataChanged := applyCodexAccountIdentityFields(clientMetadata, account, apiKeyID) + if applyCodexAccountIdentityEmbeddedMetadata(clientMetadata, account, apiKeyID) { + metadataChanged = true + } + if metadataChanged { + raw, err := json.Marshal(clientMetadata) + if err != nil { + return body, false, fmt.Errorf("encode account-scoped client_metadata: %w", err) + } + var setErr error + next, setErr = sjson.SetRawBytes(next, "client_metadata", raw) + if setErr != nil { + return body, false, fmt.Errorf("splice account-scoped client_metadata: %w", setErr) + } + changed = true + } + } + if promptCacheKey := gjson.GetBytes(body, "prompt_cache_key"); promptCacheKey.Type == gjson.String && strings.TrimSpace(promptCacheKey.String()) != "" { + raw := promptCacheKey.String() + kind := "prompt-cache" + if strings.TrimSpace(originalBodySessionID) != "" && raw == originalBodySessionID { + kind = "session" + } + scoped := scopeCodexAccountIdentityValue(account, apiKeyID, kind, raw) + if scoped != raw { + rewritten, err := sjson.SetBytes(next, "prompt_cache_key", scoped) + if err != nil { + return body, false, fmt.Errorf("splice account-scoped prompt_cache_key: %w", err) + } + next = rewritten + changed = true + } + } + return next, changed, nil +} + +func applyCodexAccountIdentityHeaders(headers http.Header, account *Account, apiKeyID int64) { + if headers == nil || codexAccountIdentityNamespace(account) == "" { + return + } + for _, field := range codexAccountIdentityFields { + // Underscore session/conversation headers are rebuilt separately from the + // prompt cache key by each request builder. + if field.name == "session_id" { + continue + } + raw := strings.TrimSpace(headers.Get(field.name)) + if raw != "" { + headers.Set(field.name, scopeCodexAccountIdentityValue(account, apiKeyID, field.kind, raw)) + } + } + if raw := strings.TrimSpace(headers.Get(openAIWSTurnMetadataHeader)); raw != "" { + metadata := map[string]any{} + if err := json.Unmarshal([]byte(raw), &metadata); err == nil && metadata != nil && applyCodexAccountIdentityFields(metadata, account, apiKeyID) { + if rebuilt, err := json.Marshal(metadata); err == nil { + headers.Set(openAIWSTurnMetadataHeader, string(rebuilt)) + } + } + } +} diff --git a/backend/internal/service/openai_codex_account_identity_test.go b/backend/internal/service/openai_codex_account_identity_test.go new file mode 100644 index 0000000000..ffceba2d39 --- /dev/null +++ b/backend/internal/service/openai_codex_account_identity_test.go @@ -0,0 +1,230 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +type codexAccountIdentityRepoStub struct { + AccountRepository + account *Account +} + +func (s *codexAccountIdentityRepoStub) GetByID(_ context.Context, _ int64) (*Account, error) { + return s.account, nil +} + +func TestCodexRequestBodyIdentityNamespaceIsStablePerOAuthAccount(t *testing.T) { + body := []byte(`{"model":"gpt-5.6-codex","prompt_cache_key":"client-session","client_metadata":{"x-codex-installation-id":"client-installation","session_id":"client-session","thread_id":"client-thread","x-codex-window-id":"client-window","x-codex-turn-metadata":"{\"installation_id\":\"client-installation\",\"session_id\":\"client-session\",\"thread_id\":\"client-thread\",\"turn_id\":\"client-turn\",\"window_id\":\"client-window\"}"}}`) + account11 := &Account{ID: 11, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "chatgpt-account-11"}} + account19 := &Account{ID: 19, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "chatgpt-account-19"}} + + first, changed, err := applyCodexAccountIdentityClientMetadataRaw(body, account11, 77) + require.NoError(t, err) + require.True(t, changed) + firstAgain, changed, err := applyCodexAccountIdentityClientMetadataRaw(body, account11, 77) + require.NoError(t, err) + require.True(t, changed) + second, changed, err := applyCodexAccountIdentityClientMetadataRaw(body, account19, 77) + require.NoError(t, err) + require.True(t, changed) + require.JSONEq(t, string(first), string(firstAgain)) + + paths := []string{ + "prompt_cache_key", + "client_metadata.x-codex-installation-id", + "client_metadata.session_id", + "client_metadata.thread_id", + "client_metadata.x-codex-window-id", + } + for _, path := range paths { + require.NotEqual(t, gjson.GetBytes(body, path).String(), gjson.GetBytes(first, path).String(), path) + require.NotEqual(t, gjson.GetBytes(first, path).String(), gjson.GetBytes(second, path).String(), path) + } + require.Equal(t, gjson.GetBytes(first, "prompt_cache_key").String(), gjson.GetBytes(first, "client_metadata.session_id").String()) + + var embeddedFirst map[string]any + var embeddedSecond map[string]any + require.NoError(t, json.Unmarshal([]byte(gjson.GetBytes(first, "client_metadata.x-codex-turn-metadata").String()), &embeddedFirst)) + require.NoError(t, json.Unmarshal([]byte(gjson.GetBytes(second, "client_metadata.x-codex-turn-metadata").String()), &embeddedSecond)) + for _, field := range []string{"installation_id", "session_id", "thread_id", "turn_id", "window_id"} { + require.NotEqual(t, embeddedFirst[field], embeddedSecond[field], field) + } +} + +func TestCodexAccountIdentityNamespaceUsesStableCredentialSource(t *testing.T) { + firstRow := &Account{ID: 8, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "shared-upstream-account"}} + secondRow := &Account{ID: 19, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "shared-upstream-account"}} + require.Equal(t, codexAccountIdentityNamespace(firstRow), codexAccountIdentityNamespace(secondRow)) + + firstUser := &Account{ID: 20, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "team-account", "chatgpt_user_id": "user-1"}} + sameUser := &Account{ID: 21, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "team-account", "chatgpt_user_id": "user-1"}} + secondUser := &Account{ID: 22, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "team-account", "chatgpt_user_id": "user-2"}} + require.Equal(t, codexAccountIdentityNamespace(firstUser), codexAccountIdentityNamespace(sameUser)) + require.NotEqual(t, codexAccountIdentityNamespace(firstUser), codexAccountIdentityNamespace(secondUser)) + + seed := "11111111-1111-4111-8111-111111111111" + seeded := &Account{ID: 11, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: map[string]any{codexFingerprintSeedExtraKey: seed}} + require.Equal(t, "seed:"+seed, codexAccountIdentityNamespace(seeded)) + + // Local row IDs repeat across independent deployments, so they are not a + // safe fallback for upstream identity. + require.Empty(t, codexAccountIdentityNamespace(&Account{ID: 11, Platform: PlatformOpenAI, Type: AccountTypeOAuth})) + + setupTokenA := &Account{ID: 30, Platform: PlatformOpenAI, Type: AccountTypeSetupToken, Credentials: map[string]any{"access_token": "setup-token-a"}} + setupTokenADuplicate := &Account{ID: 31, Platform: PlatformOpenAI, Type: AccountTypeSetupToken, Credentials: map[string]any{"access_token": "setup-token-a"}} + setupTokenB := &Account{ID: 32, Platform: PlatformOpenAI, Type: AccountTypeSetupToken, Credentials: map[string]any{"access_token": "setup-token-b"}} + setupNamespace := codexAccountIdentityNamespace(setupTokenA) + require.NotEmpty(t, setupNamespace) + require.NotContains(t, setupNamespace, "setup-token-a") + require.Equal(t, setupNamespace, codexAccountIdentityNamespace(setupTokenADuplicate)) + require.NotEqual(t, setupNamespace, codexAccountIdentityNamespace(setupTokenB)) +} + +func TestCodexAccountIdentitySourceResolvesShadowAndOverwritesFailoverContext(t *testing.T) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + + parentID := int64(11) + parent := &Account{ID: parentID, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{ + "chatgpt_account_id": "team-account", + "chatgpt_user_id": "user-1", + }} + shadow := &Account{ID: 111, ParentAccountID: &parentID, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + service := &OpenAIGatewayService{accountRepo: &codexAccountIdentityRepoStub{account: parent}} + + resolved, err := service.prepareCodexAccountIdentitySource(context.Background(), c, shadow) + require.NoError(t, err) + require.Same(t, parent, resolved) + require.Same(t, parent, codexAccountIdentitySource(c, shadow)) + + req, err := service.buildUpstreamRequest( + context.Background(), c, shadow, + []byte(`{"model":"gpt-5.6-codex","stream":true,"prompt_cache_key":"client-session"}`), + "token", true, "client-session", true, + ) + require.NoError(t, err) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, parent, "client-session"), req.Header.Get("session_id")) + + next := &Account{ID: 19, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{ + "chatgpt_account_id": "other-account", + "chatgpt_user_id": "user-2", + }} + resolved, err = service.prepareCodexAccountIdentitySource(context.Background(), c, next) + require.NoError(t, err) + require.Same(t, next, resolved) + require.Same(t, next, codexAccountIdentitySource(c, shadow)) +} + +func TestBuildOpenAIWSHeadersNamespacesCodexIdentityByOAuthAccount(t *testing.T) { + gin.SetMode(gin.TestMode) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + c.Set("api_key_id", int64(77)) + c.Request.Header.Set("x-codex-installation-id", "client-installation") + c.Request.Header.Set("thread-id", "client-thread") + c.Request.Header.Set("x-codex-window-id", "client-window") + c.Request.Header.Set("x-client-request-id", "client-request") + + account11 := &Account{ID: 11, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "chatgpt-account-11"}} + account19 := &Account{ID: 19, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Credentials: map[string]any{"chatgpt_account_id": "chatgpt-account-19"}} + service := &OpenAIGatewayService{} + build := func(account *Account) http.Header { + headers, _, err := service.buildOpenAIWSHeaders( + context.Background(), c, account, "token", + OpenAIWSProtocolDecision{Transport: OpenAIUpstreamTransportResponsesWebsocketV2}, + true, "", "", "client-session", "", "", + ) + require.NoError(t, err) + return headers + } + + first := build(account11) + firstAgain := build(account11) + second := build(account19) + for _, header := range []string{"session_id", "x-codex-installation-id", "thread-id", "x-codex-window-id", "x-client-request-id"} { + require.NotEmpty(t, first.Get(header), header) + require.Equal(t, first.Get(header), firstAgain.Get(header), header) + require.NotEqual(t, first.Get(header), second.Get(header), header) + } + + httpRequest, err := service.buildUpstreamRequest( + context.Background(), c, account11, + []byte(`{"model":"gpt-5.6-codex","stream":true,"prompt_cache_key":"client-session"}`), + "token", true, "client-session", true, + ) + require.NoError(t, err) + require.Equal(t, httpRequest.Header.Get("session_id"), first.Get("session_id"), "HTTP and WS must derive the same identity from the raw client key") +} + +func TestBuildUpstreamRequestNamespacesCodexIdentityByOAuthAccount(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := &OpenAIGatewayService{} + body := []byte(`{"model":"gpt-5.6-codex","stream":true,"prompt_cache_key":"client-session"}`) + + build := func(accountID int64, chatgptAccountID string) http.Header { + t.Helper() + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Set("api_key", &APIKey{ID: 77}) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.0") + c.Request.Header.Set("x-codex-installation-id", "client-installation") + c.Request.Header.Set("x-codex-window-id", "client-window") + c.Request.Header.Set("session-id", "client-session") + c.Request.Header.Set("thread-id", "client-thread") + c.Request.Header.Set("x-client-request-id", "client-request") + c.Request.Header.Set("x-codex-turn-metadata", `{"installation_id":"client-installation","session_id":"client-session","thread_id":"client-thread","turn_id":"client-turn","window_id":"client-window"}`) + + account := &Account{ + ID: accountID, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "chatgpt_account_id": chatgptAccountID, + }, + } + req, err := svc.buildUpstreamRequest( + context.Background(), c, account, body, "oauth-token", true, "client-session", true, + ) + require.NoError(t, err) + return req.Header + } + + first := build(11, "chatgpt-account-11") + firstAgain := build(11, "chatgpt-account-11") + second := build(19, "chatgpt-account-19") + + identityHeaders := []string{ + "x-codex-installation-id", + "x-codex-window-id", + "session-id", + "session_id", + "conversation_id", + "thread-id", + "x-client-request-id", + "x-codex-turn-metadata", + } + checked := 0 + for _, header := range identityHeaders { + if first.Get(header) == "" && second.Get(header) == "" { + continue + } + checked++ + require.NotEmpty(t, first.Get(header), header) + require.Equal(t, first.Get(header), firstAgain.Get(header), "same account must retain stable identity: %s", header) + require.NotEqual(t, first.Get(header), second.Get(header), "account failover must rotate upstream identity: %s", header) + } + require.GreaterOrEqual(t, checked, 5, "test must exercise the real outbound identity surface") +} diff --git a/backend/internal/service/openai_compat_model_test.go b/backend/internal/service/openai_compat_model_test.go index 7ccfbdeadd..b52a616f85 100644 --- a/backend/internal/service/openai_compat_model_test.go +++ b/backend/internal/service/openai_compat_model_test.go @@ -1010,7 +1010,7 @@ func TestForwardAsAnthropic_ReusesOAuthCodexTurnState(t *testing.T) { require.NoError(t, err) require.NotNil(t, secondResult) require.Equal(t, "turn_state_first", upstream.requests[1].Header.Get("x-codex-turn-state")) - require.Equal(t, generateSessionUUID(isolateOpenAISessionID(0, "stable-cache-key")), upstream.requests[1].Header.Get("session_id")) + require.Equal(t, generateSessionUUID(isolateOpenAIUpstreamSessionID(0, account, "stable-cache-key")), upstream.requests[1].Header.Get("session_id")) require.Empty(t, upstream.requests[1].Header.Get("conversation_id")) requireOpenAIMessagesCodexIdentity(t, upstream.requests[1], codexCLIUserAgent, openai.CodexDefaultOriginator) require.False(t, gjson.GetBytes(upstream.bodies[1], "prompt_cache_key").Exists()) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 5699dc09e9..69c1458fdc 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -61,6 +61,9 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) setCodexToolNameReverse(c, nil) + if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { + return nil, err + } restrictionResult := s.detectCodexClientRestriction(c, account, body) logCodexCLIOnlyDetection(ctx, c, account, getAPIKeyIDFromContext(c), restrictionResult, body) @@ -261,6 +264,7 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( } else if promptCacheKey != "" { reqBody["prompt_cache_key"] = promptCacheKey } + applyCodexAccountIdentityClientMetadataMap(reqBody, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) responsesBody, err = json.Marshal(reqBody) if err != nil { return nil, fmt.Errorf("remarshal after codex transform: %w", err) @@ -314,7 +318,7 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( if promptCacheKey != "" { apiKeyID := getAPIKeyIDFromContext(c) - upstreamReq.Header.Set("session_id", generateSessionUUID(isolateOpenAISessionID(apiKeyID, promptCacheKey))) + upstreamReq.Header.Set("session_id", generateSessionUUID(isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), promptCacheKey))) } // 7. Send request diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index eb2f5c55a6..ef53b6732d 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -24,6 +24,9 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco clearOpenAIResponsesClientToolMapping(c) clearOpenAIResponsesNamespaceNames(c) setCodexToolNameReverse(c, nil) + if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { + return nil, err + } startTime := time.Now() // 固定渠道映射后的请求级 canonical body;账号 normalize/strip 不得改写跨 failover hint。 canonicalImageIntentBody := body @@ -252,6 +255,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } bodyModified := false + clientPromptCacheKey := promptCacheKey var reqBody map[string]any ensureReqBody := func() (map[string]any, error) { if requestView.HasPatches() { @@ -478,6 +482,15 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if !isCompactRequest && applyCodexClientMetadata(decoded, account) { markDecodedModified() } + if currentClientPromptCacheKey, ok := decoded["prompt_cache_key"].(string); ok { + clientPromptCacheKey = currentClientPromptCacheKey + } + // Account namespace is orthogonal to fingerprint convergence: preserve + // each client's identity cardinality, but never reuse it across OAuth + // credentials after scheduler failover. + if !isCompactRequest && applyCodexAccountIdentityClientMetadataMap(decoded, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) { + markDecodedModified() + } stageCodexFingerprintIDs(c, nil) // 指纹收敛:一次性解析收敛 ID,请求体和出站头共享同一份 IDs(保证 turn_id 等随机字段一致)。 // fingerprintIDs 在此处解析,后续 buildUpstreamRequest 中使用同一份。 @@ -500,7 +513,13 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if codexResult.NormalizedModel != "" { upstreamModel = codexResult.NormalizedModel } - if currentPromptCacheKey, ok := decoded["prompt_cache_key"].(string); ok && currentPromptCacheKey != "" { + if strings.TrimSpace(clientPromptCacheKey) != "" { + // The body now carries an account-scoped value. Keep the original here + // so the header builder derives the same namespace exactly once. + promptCacheKey = clientPromptCacheKey + } else if currentPromptCacheKey, ok := decoded["prompt_cache_key"].(string); ok && currentPromptCacheKey != "" { + // Fingerprint convergence may inject a default key when the client did + // not provide one; preserve that existing fallback. promptCacheKey = currentPromptCacheKey } else if codexResult.PromptCacheKey != "" { promptCacheKey = codexResult.PromptCacheKey @@ -773,6 +792,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco c, account, wsReqBody, + clientPromptCacheKey, token, wsDecision, isCodexCLI, @@ -1321,12 +1341,12 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. req.Header.Set("version", CodexCanonicalClientVersion()) } compactSession := resolveOpenAICompactSessionID(c) - req.Header.Set("session_id", isolateOpenAISessionID(apiKeyID, compactSession)) + req.Header.Set("session_id", isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), compactSession)) } else { req.Header.Set("accept", "text/event-stream") } if promptCacheKey != "" { - isolated := isolateOpenAISessionID(apiKeyID, promptCacheKey) + isolated := isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), promptCacheKey) req.Header.Set("session_id", isolated) if !compatMessagesBridge || clientConversationID != "" { req.Header.Set("conversation_id", isolated) @@ -1350,6 +1370,10 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. req.Header.Set("user-agent", CodexCanonicalUserAgent()) } + // 账号 namespace 不改变客户端身份基数,但确保 scheduler failover 后不会把 + // 同一组 Codex IDs 发送给另一份 OAuth 凭据。可选指纹收敛随后仍可覆盖这些值。 + applyCodexAccountIdentityHeaders(req.Header, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + // 指纹收敛:使用 Forward() 中预计算的收敛 ID 改写出站头,与请求体使用同一份 IDs。 applyStagedCodexFingerprintHeaders(c, account, req.Header) diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 734cde2cab..17c4d633d1 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -35,6 +35,9 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( ) (*OpenAIForwardResult, error) { beginUpstreamResponseModelObservation(c) setCodexToolNameReverse(c, nil) + if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { + return nil, err + } // 入口分流(国产供应商 Anthropic 协议):上游为供应商原生 Anthropic 端点时, // /v1/messages 请求零转换直通(仅模型名映射 + 少量 body 清洗),完整保留 @@ -227,6 +230,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( if codexResult.PromptCacheKey != "" { promptCacheKey = codexResult.PromptCacheKey } + applyCodexAccountIdentityClientMetadataMap(reqBody, codexAccountIdentitySource(c, account), apiKeyID) delete(reqBody, "prompt_cache_key") if shouldAutoInjectPromptCacheKeyForCompat(upstreamModel) { compatTurnState = s.getOpenAICompatSessionTurnState(ctx, c, account, promptCacheKey) @@ -329,7 +333,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( // Override session_id with a deterministic UUID derived from the isolated // session key, ensuring different API keys produce different upstream sessions. if account.Platform != PlatformGrok && promptCacheKey != "" { - isolatedSessionID := generateSessionUUID(isolateOpenAISessionID(apiKeyID, promptCacheKey)) + isolatedSessionID := generateSessionUUID(isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), promptCacheKey)) upstreamReq.Header.Set("session_id", isolatedSessionID) if upstreamReq.Header.Get("conversation_id") != "" { upstreamReq.Header.Set("conversation_id", isolatedSessionID) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index b2115357d7..d5d0c75408 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -180,6 +180,14 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( } reqStream = gjson.GetBytes(body, "stream").Bool() + accountScopedBody, accountScoped, scopeErr := applyCodexAccountIdentityClientMetadataRaw(body, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + if scopeErr != nil { + return nil, scopeErr + } + if accountScoped { + body = accountScopedBody + } + stageCodexFingerprintIDs(c, nil) // 指纹收敛:与非透传路径同门控(仅 OAuth、legacy compact 形态跳过)。 // 一次性解析收敛 ID:请求体 client_metadata 在此改写(raw 字节外科 @@ -670,10 +678,10 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( clientConversationID = promptCacheKey } if clientSessionID != "" { - req.Header.Set("session_id", isolateOpenAISessionID(apiKeyID, clientSessionID)) + req.Header.Set("session_id", isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), clientSessionID)) } if clientConversationID != "" { - req.Header.Set("conversation_id", isolateOpenAISessionID(apiKeyID, clientConversationID)) + req.Header.Set("conversation_id", isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), clientConversationID)) } } else if isOpenAIResponsesCompactPath(c) { // 透传白名单会放行客户端的 Accept: text/event-stream;compact 上游是 @@ -690,6 +698,8 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( if s.cfg != nil && s.cfg.Gateway.ForceCodexCLI { req.Header.Set("user-agent", CodexCanonicalUserAgent()) } + applyCodexAccountIdentityHeaders(req.Header, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + // 指纹收敛:使用 forwardOpenAIPassthrough 中预计算的收敛 ID 改写出站头, // 与请求体 client_metadata 共享同一份 IDs(与非透传路径相同的相对位置: // 会话隔离之后、终态身份收口之前)。 diff --git a/backend/internal/service/openai_setup_token_compat_test.go b/backend/internal/service/openai_setup_token_compat_test.go index 7405deb419..8f2f2bfcee 100644 --- a/backend/internal/service/openai_setup_token_compat_test.go +++ b/backend/internal/service/openai_setup_token_compat_test.go @@ -227,7 +227,7 @@ func TestOpenAISetupTokenMessagesUsesCodexBridgeAndTurnState(t *testing.T) { require.NotNil(t, secondResult) require.True(t, isOpenAICompatMessagesBridgeContext(secondCtx)) require.Equal(t, "turn_state_setup", upstream.requests[1].Header.Get("x-codex-turn-state")) - require.Equal(t, generateSessionUUID(isolateOpenAISessionID(0, "stable-cache-key")), upstream.requests[1].Header.Get("session_id")) + require.Equal(t, generateSessionUUID(isolateOpenAIUpstreamSessionID(0, account, "stable-cache-key")), upstream.requests[1].Header.Get("session_id")) require.Empty(t, upstream.requests[1].Header.Get("conversation_id")) requireOpenAIMessagesCodexIdentity(t, upstream.requests[1], codexCLIUserAgent, "codex-tui") } diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index 0265ad6903..5eff93fc2e 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -86,6 +86,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( // A handler may reuse the same gin context across account failover attempts. // Never let an OAuth attempt's response aliases leak into the next account. setCodexToolNameReverse(c, nil) + if _, err := s.prepareCodexAccountIdentitySource(ctx, c, account); err != nil { + return err + } if err := validateOpenAIWSBearerToken(account, token); err != nil { return err } @@ -181,15 +184,16 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( isCodexCLI := openai.IsCodexOfficialClientByHeaders(c.GetHeader("User-Agent"), c.GetHeader("originator")) || (s.cfg != nil && s.cfg.Gateway.ForceCodexCLI) type openAIWSClientPayload struct { - payloadRaw []byte - rawForHash []byte - promptCacheKey string - previousResponseID string - originalModel string - imageBillingModel string - imageSizeTier string - imageInputSize string - payloadBytes int + payloadRaw []byte + accountIdentitySourceRaw []byte + rawForHash []byte + promptCacheKey string + previousResponseID string + originalModel string + imageBillingModel string + imageSizeTier string + imageInputSize string + payloadBytes int } ingressSessionOriginalModel := "" @@ -308,6 +312,14 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } normalized = next } + accountIdentitySourceRaw := append([]byte(nil), normalized...) + accountScopedPayload, accountScoped, scopeErr := applyCodexAccountIdentityClientMetadataRaw(normalized, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + if scopeErr != nil { + return openAIWSClientPayload{}, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket identity metadata", scopeErr) + } + if accountScoped { + normalized = accountScopedPayload + } if account.IsOpenAIOAuthLike() && isOpenAIResponsesLiteWebSocketPayload(normalized) { litePayload, _, liteErr := normalizeOpenAIResponsesLiteToolsPayload(normalized) if liteErr != nil { @@ -450,15 +462,16 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( ingressSessionOriginalModel = originalModel return openAIWSClientPayload{ - payloadRaw: normalized, - rawForHash: trimmed, - promptCacheKey: promptCacheKey, - previousResponseID: previousResponseID, - originalModel: originalModel, - imageBillingModel: imageBillingModel, - imageSizeTier: imageSizeTier, - imageInputSize: imageInputSize, - payloadBytes: len(normalized), + payloadRaw: normalized, + accountIdentitySourceRaw: accountIdentitySourceRaw, + rawForHash: trimmed, + promptCacheKey: promptCacheKey, + previousResponseID: previousResponseID, + originalModel: originalModel, + imageBillingModel: imageBillingModel, + imageSizeTier: imageSizeTier, + imageInputSize: imageInputSize, + payloadBytes: len(normalized), }, nil } @@ -645,7 +658,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( var failoverErr *UpstreamFailoverError if turn > 1 && errors.As(bridgeErr, &failoverErr) && failoverErr != nil { retryPayload, retrySafe, retryPayloadErr := buildOpenAIWSCurrentTurnRetryPayload( - bridgePayloadRaw, + currentBridgePayload.accountIdentitySourceRaw, turnAccountFailoverInput, turnAccountFailoverInputExists, currentBridgePayload.originalModel, diff --git a/backend/internal/service/openai_ws_forwarder_ingress_session_test.go b/backend/internal/service/openai_ws_forwarder_ingress_session_test.go index 7633c9061e..097d845037 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress_session_test.go +++ b/backend/internal/service/openai_ws_forwarder_ingress_session_test.go @@ -1242,7 +1242,7 @@ func TestOpenAIGatewayService_ProxyResponsesWebSocketFromClient_PassthroughHeade t.Fatal("等待 passthrough websocket 结束超时") } - require.Equal(t, isolateOpenAISessionID(0, "pcache_passthrough"), captureDialer.lastHeaders.Get("session_id")) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "pcache_passthrough"), captureDialer.lastHeaders.Get("session_id")) require.Equal(t, "turn-state-1", captureDialer.lastHeaders.Get(openAIWSTurnStateHeader)) require.Equal(t, "turn-meta-1", captureDialer.lastHeaders.Get(openAIWSTurnMetadataHeader)) require.Len(t, upstreamConn.writes, 1) diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index 89d0dfc18c..94738221a2 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -121,10 +121,10 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( if account != nil && account.UsesOpenAICodexProtocol() { apiKeyID := getAPIKeyIDFromContext(c) if sessionResolution.SessionID != "" { - headers.Set("session_id", isolateOpenAISessionID(apiKeyID, sessionResolution.SessionID)) + headers.Set("session_id", isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), sessionResolution.SessionID)) } if sessionResolution.ConversationID != "" { - headers.Set("conversation_id", isolateOpenAISessionID(apiKeyID, sessionResolution.ConversationID)) + headers.Set("conversation_id", isolateOpenAIUpstreamSessionID(apiKeyID, codexAccountIdentitySource(c, account), sessionResolution.ConversationID)) } } else { if sessionResolution.SessionID != "" { @@ -140,6 +140,7 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( if metadata := strings.TrimSpace(turnMetadata); metadata != "" { headers.Set(openAIWSTurnMetadataHeader, metadata) } + applyCodexAccountIdentityHeaders(headers, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) applyStagedCodexFingerprintHeaders(c, account, headers) if account != nil && account.UsesOpenAICodexProtocol() { diff --git a/backend/internal/service/openai_ws_forwarder_success_test.go b/backend/internal/service/openai_ws_forwarder_success_test.go index 24f23f0d2b..e775104fa2 100644 --- a/backend/internal/service/openai_ws_forwarder_success_test.go +++ b/backend/internal/service/openai_ws_forwarder_success_test.go @@ -430,7 +430,7 @@ func TestOpenAIGatewayService_BuildOpenAIWSHeadersPreservesCodexIdentity(t *test require.Empty(t, headers.Get("X-Test")) } -func TestOpenAIGatewayService_BuildOpenAIWSHeadersDeviceModePreservesClientSessionIdentity(t *testing.T) { +func TestOpenAIGatewayService_BuildOpenAIWSHeadersDeviceModePreservesNamespacedClientSessionIdentity(t *testing.T) { gin.SetMode(gin.TestMode) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) @@ -465,10 +465,10 @@ func TestOpenAIGatewayService_BuildOpenAIWSHeadersDeviceModePreservesClientSessi require.NoError(t, err) require.Equal(t, ids.installationID, headers.Get("x-codex-installation-id")) require.NotEqual(t, "client-installation", headers.Get("x-codex-installation-id")) - require.Equal(t, "client-window", headers.Get("x-codex-window-id")) - require.Equal(t, "client-session", headers.Get("session-id")) - require.Equal(t, "client-thread", headers.Get("thread-id")) - require.Equal(t, "client-request", headers.Get("x-client-request-id")) + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "window", "client-window"), headers.Get("x-codex-window-id")) + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "session", "client-session"), headers.Get("session-id")) + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "thread", "client-thread"), headers.Get("thread-id")) + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "request", "client-request"), headers.Get("x-client-request-id")) } func TestLogOpenAIWSBindResponseAccountWarn(t *testing.T) { @@ -798,10 +798,10 @@ func TestOpenAIGatewayService_Forward_WSv2_OAuthStoreFalseByDefault(t *testing.T require.Equal(t, "native-wsv2", gjson.Get(requestJSON, "input.0.namespace").String(), "OAuth WSv2 应保留原生 namespace") require.Equal(t, openAIWSBetaV2Value, captureDialer.lastHeaders.Get("OpenAI-Beta")) require.Equal(t, "remote_compaction_v2", captureDialer.lastHeaders.Get("x-codex-beta-features")) - // OAuth 账号的 session_id/conversation_id 应被 isolateOpenAISessionID 隔离, + // OAuth 账号的 session_id/conversation_id 应同时按 API key 和上游账号隔离, // 测试中未设置 api_key 到 context,apiKeyID=0。 - require.Equal(t, isolateOpenAISessionID(0, "sess-oauth-1"), captureDialer.lastHeaders.Get("session_id")) - require.Equal(t, isolateOpenAISessionID(0, "conv-oauth-1"), captureDialer.lastHeaders.Get("conversation_id")) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "sess-oauth-1"), captureDialer.lastHeaders.Get("session_id")) + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "conv-oauth-1"), captureDialer.lastHeaders.Get("conversation_id")) } func TestOpenAIGatewayService_Forward_WSv2_OAuthOriginatorCompatibility(t *testing.T) { @@ -1023,8 +1023,8 @@ func TestOpenAIGatewayService_Forward_WSv2_HeaderSessionFallbackFromPromptCacheK require.NotNil(t, result) require.Equal(t, "resp_prompt_cache_key", result.RequestID) - // OAuth 账号的 session_id 应被 isolateOpenAISessionID 隔离(apiKeyID=0,未在 context 设置)。 - require.Equal(t, isolateOpenAISessionID(0, "pcache_123"), captureDialer.lastHeaders.Get("session_id")) + // OAuth 账号的 session_id 应同时按 API key 和上游账号隔离(apiKeyID=0)。 + require.Equal(t, isolateOpenAIUpstreamSessionID(0, account, "pcache_123"), captureDialer.lastHeaders.Get("session_id")) require.Empty(t, captureDialer.lastHeaders.Get("conversation_id")) require.NotNil(t, captureConn.lastWrite) require.True(t, gjson.Get(requestToJSONString(captureConn.lastWrite), "stream").Exists()) diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 06af82c7fb..7702ebe797 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -22,6 +22,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( c *gin.Context, account *Account, reqBody map[string]any, + clientPromptCacheKey string, token string, decision OpenAIWSProtocolDecision, isCodexCLI bool, @@ -72,7 +73,12 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( applyStagedCodexFingerprintClientMetadata(c, account, payload) previousResponseID := openAIWSPayloadString(payload, "previous_response_id") previousResponseIDKind := ClassifyOpenAIPreviousResponseIDKind(previousResponseID) - promptCacheKey := openAIWSPayloadString(payload, "prompt_cache_key") + promptCacheKey := strings.TrimSpace(clientPromptCacheKey) + if promptCacheKey == "" { + // Fingerprint convergence may inject a default key when the client did + // not send one; retain that fallback without replacing an explicit raw key. + promptCacheKey = openAIWSPayloadString(payload, "prompt_cache_key") + } _, hasTools := payload["tools"] debugEnabled := isOpenAIWSModeDebugEnabled() payloadBytes := -1 diff --git a/backend/internal/service/openai_ws_http_bridge_resume_test.go b/backend/internal/service/openai_ws_http_bridge_resume_test.go index 074c08f8ce..1468892bbe 100644 --- a/backend/internal/service/openai_ws_http_bridge_resume_test.go +++ b/backend/internal/service/openai_ws_http_bridge_resume_test.go @@ -150,11 +150,13 @@ func TestOpenAIWSHTTPBridgeLaterTurn429RetriesCurrentTurnOnReplacementAccount(t account := &Account{ ID: 129, Name: "limited", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive, Schedulable: true, Concurrency: 1, - Extra: map[string]any{"openai_oauth_responses_websockets_v2_mode": OpenAIWSIngressModeHTTPBridge}, + Extra: map[string]any{"openai_oauth_responses_websockets_v2_mode": OpenAIWSIngressModeHTTPBridge}, + Credentials: map[string]any{"chatgpt_account_id": "account-a", "chatgpt_user_id": "user-a"}, } nextAccount := *account nextAccount.ID = 130 nextAccount.Name = "replacement" + nextAccount.Credentials = map[string]any{"chatgpt_account_id": "account-b", "chatgpt_user_id": "user-b"} serverErrCh := make(chan error, 1) failoverCh := make(chan []byte, 1) @@ -212,7 +214,7 @@ func TestOpenAIWSHTTPBridgeLaterTurn429RetriesCurrentTurnOnReplacementAccount(t require.Equal(t, "response.completed", gjson.GetBytes(completed, "type").String()) writeCtx, cancel = context.WithTimeout(context.Background(), 3*time.Second) - err = clientConn.Write(writeCtx, websocket.MessageText, []byte(`{"type":"response.create","model":"gpt-5.6-sol","previous_response_id":"resp_first","input":[{"type":"function_call_output","call_id":"call_1","output":"second"}]}`)) + err = clientConn.Write(writeCtx, websocket.MessageText, []byte(`{"type":"response.create","model":"gpt-5.6-sol","previous_response_id":"resp_first","prompt_cache_key":"client-session","client_metadata":{"session_id":"client-session","thread_id":"client-thread"},"input":[{"type":"function_call_output","call_id":"call_1","output":"second"}]}`)) cancel() require.NoError(t, err) @@ -237,6 +239,8 @@ func TestOpenAIWSHTTPBridgeLaterTurn429RetriesCurrentTurnOnReplacementAccount(t require.Contains(t, input.Raw, "second") require.Equal(t, 1, strings.Count(input.Raw, `"id":"fc_1"`)) require.Equal(t, 2, strings.Count(input.Raw, `"call_id":"call_1"`)) + require.Equal(t, "client-session", gjson.GetBytes(retryPayload, "client_metadata.session_id").String()) + require.Equal(t, "client-thread", gjson.GetBytes(retryPayload, "client_metadata.thread_id").String()) case <-time.After(5 * time.Second): t.Fatal("timed out waiting for current-turn failover") } @@ -249,7 +253,11 @@ func TestOpenAIWSHTTPBridgeLaterTurn429RetriesCurrentTurnOnReplacementAccount(t } require.Len(t, upstream.bodies, 3) require.Contains(t, string(upstream.bodies[0]), "first") + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "session", "client-session"), gjson.GetBytes(upstream.bodies[1], "client_metadata.session_id").String()) + require.Equal(t, scopeCodexAccountIdentityValue(account, 0, "thread", "client-thread"), gjson.GetBytes(upstream.bodies[1], "client_metadata.thread_id").String()) require.NotContains(t, string(upstream.bodies[2]), "previous_response_id") require.Contains(t, string(upstream.bodies[2]), "second") + require.Equal(t, scopeCodexAccountIdentityValue(&nextAccount, 0, "session", "client-session"), gjson.GetBytes(upstream.bodies[2], "client_metadata.session_id").String()) + require.Equal(t, scopeCodexAccountIdentityValue(&nextAccount, 0, "thread", "client-thread"), gjson.GetBytes(upstream.bodies[2], "client_metadata.thread_id").String()) require.Empty(t, upstream.requests[2].Header.Get(openAIWSTurnStateHeader)) } diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index c8fcd87525..96e1c859bb 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -693,6 +693,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( } requestModel := strings.TrimSpace(gjson.GetBytes(firstClientMessage, "model").String()) requestPreviousResponseID := strings.TrimSpace(gjson.GetBytes(firstClientMessage, "previous_response_id").String()) + promptCacheKey := strings.TrimSpace(gjson.GetBytes(firstClientMessage, "prompt_cache_key").String()) logOpenAIWSV2Passthrough( "relay_start account_id=%d model=%s previous_response_id=%s first_message_type=%s first_message_bytes=%d", account.ID, @@ -749,6 +750,13 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( firstClientMessage = aliasedBody } } + accountScopedFirst, accountScoped, scopeErr := applyCodexAccountIdentityClientMetadataRaw(firstClientMessage, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + if scopeErr != nil { + return NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket identity metadata", scopeErr) + } + if accountScoped { + firstClientMessage = accountScopedFirst + } usageMeta := newOpenAIWSPassthroughUsageMeta(initialRequestModel, firstClientMessage) updatedFirst, blocked, policyErr := s.applyOpenAIFastPolicyToWSResponseCreate(ctx, account, capturedSessionModel, firstClientMessage) if policyErr != nil { @@ -789,8 +797,6 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( usageMeta.initFromFirstFrame(firstClientMessage, capturedSessionModel) _, initialUpstreamModel := usageMeta.turnModels(initialRequestModel) SetOpsUpstreamModel(c, initialUpstreamModel) - promptCacheKey := strings.TrimSpace(gjson.GetBytes(firstClientMessage, "prompt_cache_key").String()) - wsURL, err := s.buildOpenAIResponsesWSURL(account) if err != nil { return fmt.Errorf("build ws url: %w", err) @@ -995,6 +1001,15 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( payload = aliasedBody } } + if isResponseCreate || eventType == "session.update" { + accountScopedPayload, accountScoped, scopeErr := applyCodexAccountIdentityClientMetadataRaw(payload, codexAccountIdentitySource(c, account), getAPIKeyIDFromContext(c)) + if scopeErr != nil { + return payload, nil, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket identity metadata", scopeErr) + } + if accountScoped { + payload = accountScopedPayload + } + } if isResponseCreate { if account.IsOpenAIOAuthLike() && isOpenAIResponsesLiteWebSocketPayload(payload) { litePayload, _, liteErr := normalizeOpenAIResponsesLiteToolsPayload(payload)