fix(subscription): 修正 Codex 与 Claude 凭据身份判重 (#639)

* fix(subscription): 修正订阅凭据身份与刷新校验 * docs: 撤回 README 中的凭据身份说明 * fix(subscription): 校验令牌身份冲突并补齐刷新后判重 * fix(subscription): 防止刷新丢失已确认身份 * fix(subscription): 延后批量导入身份判重
2026-09-13 12:26:46 +08:00
1 parent f8a04df46d
commit 9605ed208b
@@ -181,15 +181,16 @@ func (s *Service) ImportCredentialFiles(
 				item.ErrorCode = credentialImportItemError(ctx, classifyCredentialImportError(driver, importErr))
 				break
 			}
-			if _, duplicate := identities[credential.Identity()]; duplicate {
-				item.Status, item.ErrorCode = "skipped", "duplicate_account"
-				break
-			}
+			// 刷新可能补全用户或组织身份，必须按准备完成后的身份判重。
 			credential, importErr = s.prepareTransientSubscriptionCredential(ctx, channelID, driver, credential)
 			if importErr != nil {
 				item.ErrorCode = credentialImportItemError(ctx, importErr)
 				break
 			}
+			if _, duplicate := identities[credential.Identity()]; duplicate {
+				item.Status, item.ErrorCode = "skipped", "duplicate_account"
+				break
+			}
 			stage, persistErr := s.persistReadyCredentialStage(ctx, channelID, "oauth_file", credential, network)
 			if persistErr != nil {
 				item.ErrorCode = credentialImportItemError(ctx, persistErr)
@@ -297,7 +297,7 @@ func (s *Service) prepareTransientSubscriptionCredential(
 		}
 		return subscriptionruntime.Credential{}, app_errors.ErrCredentialAuthOutcomeUnknown
 	}
-	if refreshed.Identity() == "" || refreshed.Identity() != credential.Identity() {
+	if !subscriptionruntime.RefreshPreservesIdentity(driver, credential, refreshed) {
 		return subscriptionruntime.Credential{}, app_errors.ErrCredentialReauthorizationRequired
 	}
 	return refreshed, nil
@@ -393,8 +393,8 @@ func (s *Service) prepareReadySubscriptionStageCredential(
 		}
 		return subscriptionruntime.Credential{}, apiErr
 	}
-	if refreshed.Identity() == "" || refreshed.Identity() != credential.Identity() ||
-		s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), refreshed.Identity()) != row.IdentityFingerprint {
+	if !subscriptionruntime.RefreshPreservesIdentity(driver, credential, refreshed) ||
+		s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), credential.Identity()) != row.IdentityFingerprint {
 		if err := s.finishCredentialStageRefreshFailure(
 			ctx,
 			row.ID,
@@ -496,7 +496,7 @@ func (s *Service) finishCredentialStageRefresh(
 			"status": models.CredentialStageReady, "encrypted_payload": ciphertext,
 			"payload_schema_version": stagedSubscriptionSchemaV2,
 			"safe_summary_json":      models.JSON(summaryJSON),
-			"identity_fingerprint":   row.IdentityFingerprint,
+			"identity_fingerprint":   s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), credential.Identity()),
 			"expires_at_ms":          row.ExpiresAtMS,
 			"error_code":             "", "updated_at_ms": s.now().UnixMilli(),
 		})
@@ -0,0 +1,277 @@
+package control
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"gpt-load/internal/channel"
+	app_errors "gpt-load/internal/platform/errors"
+	"gpt-load/internal/storage/models"
+	subscriptionruntime "gpt-load/internal/subscription/runtime"
+)
+
+func subscriptionIdentityJSON(t *testing.T, channelID channel.ID, account, scope, token, expires string) []byte {
+	t.Helper()
+	values := map[string]string{"type": string(channelID), "access_token": "access-" + token, "refresh_token": "refresh-" + token, "email": token + "@example.com", "expired": expires}
+	if channelID == channel.Codex {
+		values["account_id"] = account
+		if scope != "" {
+			claims, err := json.Marshal(map[string]any{"https://api.openai.com/auth": map[string]string{"chatgpt_account_id": account, "chatgpt_user_id": scope}})
+			if err != nil {
+				t.Fatal(err)
+			}
+			values["id_token"] = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".signature"
+		}
+	} else {
+		values["account_uuid"] = account
+		if scope != "" {
+			values["organization_uuid"] = scope
+		}
+	}
+	raw, err := json.Marshal(values)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+func TestSubscriptionConnectionUsesCompleteIdentity(t *testing.T) {
+	t.Parallel()
+	for _, test := range []struct {
+		name           string
+		channel        channel.ID
+		account, scope string
+		added          int
+	}{
+		{"codex different users", channel.Codex, "account-one", "scope-two", 1},
+		{"codex different workspaces", channel.Codex, "account-two", "scope-one", 1},
+		{"codex reauthorization", channel.Codex, "account-one", "scope-one", 0},
+		{"claude different organizations", channel.Claude, "account-one", "scope-two", 1},
+		{"claude different users", channel.Claude, "account-two", "scope-one", 1},
+		{"claude reauthorization", channel.Claude, "account-one", "scope-one", 0},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			for _, authState := range []models.CredentialAuthState{models.CredentialAuthStateReady, models.CredentialAuthStateReauthorizationRequired} {
+				t.Run(string(authState), func(t *testing.T) {
+					f := newServiceFixture(t)
+					importStage := func(account, scope, token string) CredentialStageResult {
+						result, err := f.service.ImportCredentialStage(t.Context(), test.channel, subscriptionIdentityJSON(t, test.channel, account, scope, token, "2099-01-01T00:00:00Z"))
+						if err != nil {
+							t.Fatal(err)
+						}
+						return result
+					}
+					first := importStage("account-one", "scope-one", "first")
+					group, err := f.service.CreateGroup(t.Context(), GroupCreateRequest{Name: stringPointer("identity"), ChannelID: test.channel, ConnectionType: models.ConnectionTypeSubscription, Models: optionalGroupModels{Set: true, Values: []GroupModel{}}, StagedCredentialIDs: []string{first.StageID}})
+					if err != nil {
+						t.Fatal(err)
+					}
+					var original models.Credential
+					if err := f.db.Where("group_id = ?", group.GroupID).Take(&original).Error; err != nil {
+						t.Fatal(err)
+					}
+					if err := f.db.Model(&original).Update("auth_state", authState).Error; err != nil {
+						t.Fatal(err)
+					}
+					second := importStage(test.account, test.scope, "second")
+					result, err := f.service.ConnectGroupCredentials(t.Context(), group.GroupID, []string{second.StageID})
+					if err != nil {
+						t.Fatal(err)
+					}
+					wantAdded := test.added
+					if authState == models.CredentialAuthStateReauthorizationRequired {
+						wantAdded = 1
+					}
+					if result.CredentialsAdded != wantAdded || result.CredentialsDuplicated != 1-wantAdded {
+						t.Fatalf("added/duplicated = %d/%d, want %d/%d", result.CredentialsAdded, result.CredentialsDuplicated, wantAdded, 1-wantAdded)
+					}
+					var rows []models.Credential
+					if err := f.db.Where("group_id = ?", group.GroupID).Order("id ASC").Find(&rows).Error; err != nil {
+						t.Fatal(err)
+					}
+					if len(rows) != 1+test.added {
+						t.Fatalf("credential count = %d, want %d", len(rows), 1+test.added)
+					}
+					if test.added == 1 && rows[0].Data != original.Data {
+						t.Fatal("another identity replaced the existing credential")
+					}
+				})
+			}
+		})
+	}
+}
+
+func TestSubscriptionRefreshIdentityAllowsEnrichmentWithoutDowngrade(t *testing.T) {
+	t.Parallel()
+	for _, channelID := range []channel.ID{channel.Codex, channel.Claude} {
+		t.Run(string(channelID), func(t *testing.T) {
+			for _, test := range []struct {
+				name, oldScope, newAccount, newScope string
+				allowed                              bool
+			}{
+				{"same identity", "scope-one", "account-one", "scope-one", true},
+				{"legacy metadata still absent", "", "account-one", "", true},
+				{"enriched metadata", "", "account-one", "scope-one", true},
+				{"missing metadata", "scope-one", "account-one", "", false},
+				{"different scope", "scope-one", "account-one", "scope-two", false},
+				{"different account", "scope-one", "account-two", "scope-one", false},
+			} {
+				t.Run(test.name, func(t *testing.T) {
+					for _, path := range []string{"transient", "ready stage"} {
+						t.Run(path, func(t *testing.T) {
+							f := newServiceFixture(t)
+							driver, ok := subscriptionsDriver(f.service.subscriptions, channelID)
+							if !ok {
+								t.Fatal("driver missing")
+							}
+							current, err := driver.Parse(subscriptionIdentityJSON(t, channelID, "account-one", test.oldScope, "current", "2000-01-01T00:00:00Z"))
+							if err != nil {
+								t.Fatal(err)
+							}
+							refreshed, err := driver.Parse(subscriptionIdentityJSON(t, channelID, test.newAccount, test.newScope, "next", "2099-01-01T00:00:00Z"))
+							if err != nil {
+								t.Fatal(err)
+							}
+							f.service.refreshSubscriptionCredential = func(context.Context, channel.ID, subscriptionruntime.Credential) (subscriptionruntime.Credential, error) {
+								return refreshed, nil
+							}
+							if path == "transient" {
+								_, err = f.service.prepareTransientSubscriptionCredential(t.Context(), channelID, driver, current)
+							} else {
+								stage, stageErr := f.service.persistReadyCredentialStage(t.Context(), channelID, "oauth_file", current)
+								if stageErr != nil {
+									t.Fatal(stageErr)
+								}
+								row, loadErr := f.service.loadCredentialStage(t.Context(), stage.StageID)
+								if loadErr != nil {
+									t.Fatal(loadErr)
+								}
+								_, err = f.service.prepareReadySubscriptionStageCredential(t.Context(), row, driver, current, true)
+								if test.allowed && err == nil {
+									// 正常刷新后的短期暂存必须仍可消费，不能留下过期的身份指纹。
+									_, err = f.service.CreateGroup(t.Context(), GroupCreateRequest{Name: stringPointer("refreshed identity"), ChannelID: channelID, ConnectionType: models.ConnectionTypeSubscription, Models: optionalGroupModels{Set: true, Values: []GroupModel{}}, StagedCredentialIDs: []string{stage.StageID}})
+								}
+							}
+							if test.allowed && err != nil {
+								t.Fatalf("same authorization with optional metadata change rejected: %v", err)
+							}
+							if !test.allowed && !errors.Is(err, app_errors.ErrCredentialReauthorizationRequired) {
+								t.Fatalf("changed identity error = %v, want reauthorization required", err)
+							}
+						})
+					}
+				})
+			}
+		})
+	}
+}
+
+func TestSubscriptionBatchImportSeparatesIdentityScopes(t *testing.T) {
+	t.Parallel()
+	for _, channelID := range []channel.ID{channel.Codex, channel.Claude} {
+		t.Run(string(channelID), func(t *testing.T) {
+			f := newServiceFixture(t)
+			items := []json.RawMessage{
+				subscriptionIdentityJSON(t, channelID, "account", "scope-one", "first", "2099-01-01T00:00:00Z"),
+				subscriptionIdentityJSON(t, channelID, "account", "scope-two", "second", "2099-01-01T00:00:00Z"),
+				subscriptionIdentityJSON(t, channelID, "account", "scope-one", "renewed", "2099-01-01T00:00:00Z"),
+			}
+			raw, err := json.Marshal(items)
+			if err != nil {
+				t.Fatal(err)
+			}
+			result, err := f.service.ImportCredentialBatch(t.Context(), channelID, raw, 0, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(result.Items) != 3 || result.Items[0].Status != "ready" || result.Items[1].Status != "ready" || result.Items[2].ErrorCode != "duplicate_account" {
+				t.Fatal("batch import did not distinguish scopes and deduplicate the same identity")
+			}
+		})
+	}
+}
+
+func TestSubscriptionBatchDeduplicatesIdentityAfterRefresh(t *testing.T) {
+	t.Parallel()
+	for _, ch := range []channel.ID{channel.Codex, channel.Claude} {
+		t.Run(string(ch), func(t *testing.T) {
+			f := newServiceFixture(t)
+			driver, ok := subscriptionsDriver(f.service.subscriptions, ch)
+			if !ok {
+				t.Fatal("driver missing")
+			}
+			f.service.refreshSubscriptionCredential = func(_ context.Context, _ channel.ID, current subscriptionruntime.Credential) (subscriptionruntime.Credential, error) {
+				var fields map[string]any
+				if err := json.Unmarshal(current.Canonical(), &fields); err != nil {
+					t.Fatal(err)
+				}
+				return driver.Parse(subscriptionIdentityJSON(t, ch, "account", "scope", fields["refresh_token"].(string), "2099-01-01T00:00:00Z"))
+			}
+			raw, err := json.Marshal([]json.RawMessage{
+				subscriptionIdentityJSON(t, ch, "account", "", "one", "2000-01-01T00:00:00Z"),
+				subscriptionIdentityJSON(t, ch, "account", "", "two", "2000-01-01T00:00:00Z"),
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			result, err := f.service.ImportCredentialBatch(t.Context(), ch, raw, 0, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(result.Items) != 2 || result.Items[0].Status != "ready" || result.Items[1].Status != "skipped" || result.Items[1].ErrorCode != "duplicate_account" {
+				t.Fatal("batch did not deduplicate the identity enriched by refresh")
+			}
+			var count int64
+			if err := f.db.Model(&models.CredentialStage{}).Count(&count).Error; err != nil {
+				t.Fatal(err)
+			}
+			if count != 1 {
+				t.Fatalf("persisted stages = %d, want 1", count)
+			}
+		})
+	}
+}
+
+func TestSubscriptionBatchRefreshesBeforeCoarseIdentityDeduplication(t *testing.T) {
+	t.Parallel()
+	for _, ch := range []channel.ID{channel.Codex, channel.Claude} {
+		t.Run(string(ch), func(t *testing.T) {
+			f := newServiceFixture(t)
+			driver, ok := subscriptionsDriver(f.service.subscriptions, ch)
+			if !ok {
+				t.Fatal("driver missing")
+			}
+			f.service.refreshSubscriptionCredential = func(_ context.Context, _ channel.ID, current subscriptionruntime.Credential) (subscriptionruntime.Credential, error) {
+				var fields map[string]any
+				if err := json.Unmarshal(current.Canonical(), &fields); err != nil {
+					t.Fatal(err)
+				}
+				return driver.Parse(subscriptionIdentityJSON(t, ch, "account", "scope", fields["refresh_token"].(string), "2099-01-01T00:00:00Z"))
+			}
+			raw, err := json.Marshal([]json.RawMessage{
+				subscriptionIdentityJSON(t, ch, "account", "", "one", "2099-01-01T00:00:00Z"),
+				subscriptionIdentityJSON(t, ch, "account", "", "two", "2000-01-01T00:00:00Z"),
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			result, err := f.service.ImportCredentialBatch(t.Context(), ch, raw, 0, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(result.Items) != 2 || result.Items[0].Status != "ready" || result.Items[1].Status != "ready" {
+				t.Fatal("batch skipped a credential before refresh could complete its identity")
+			}
+			var count int64
+			if err := f.db.Model(&models.CredentialStage{}).Count(&count).Error; err != nil {
+				t.Fatal(err)
+			}
+			if count != 2 {
+				t.Fatalf("persisted stages = %d, want 2", count)
+			}
+		})
+	}
+}
@@ -0,0 +1,81 @@
+package loader_test
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"gpt-load/internal/channel"
+	"gpt-load/internal/platform/encryption"
+	"gpt-load/internal/state"
+	"gpt-load/internal/state/loader"
+	"gpt-load/internal/storage/models"
+)
+
+func TestLoadPreservesLegacySubscriptionDataAndIdentity(t *testing.T) {
+	t.Parallel()
+	token := "e30." + base64.RawURLEncoding.EncodeToString([]byte(`{"https://api.openai.com/auth":{"chatgpt_account_id":"account","chatgpt_user_id":"scope"}}`)) + ".signature"
+	for _, test := range []struct {
+		channel   channel.ID
+		canonical string
+	}{
+		{channel.Codex, fmt.Sprintf(`{"type":"codex","id_token":%q,"access_token":"access","refresh_token":"refresh","account_id":"account"}`, token)},
+		{channel.Claude, fmt.Sprintf(`{"type":"claude","access_token":"access","refresh_token":"refresh","account_uuid":"account","organization_uuid":"scope","claude_device_ids":[%q],"expired":"2099-01-01T00:00:00Z"}`, strings.Repeat("a", 64))},
+	} {
+		t.Run(string(test.channel), func(t *testing.T) {
+			db := openMigratedDatabase(t)
+			group := models.Group{Name: "legacy identity", ChannelID: string(test.channel), ConnectionType: models.ConnectionTypeSubscription, Params: models.JSON(`{}`), Models: models.JSON(`[]`), Overrides: models.JSON(`{}`), Enabled: true}
+			mustCreate(t, db, &group)
+			crypto, err := encryption.NewService("legacy-identity-test-key-material")
+			if err != nil {
+				t.Fatal(err)
+			}
+			ciphertext, err := crypto.Encrypt(test.canonical)
+			if err != nil {
+				t.Fatal(err)
+			}
+			legacyIdentity := crypto.Hash("credential-identity/v1|" + string(test.channel) + "|" + string(test.channel) + "|account")
+			row := models.Credential{GroupID: group.ID, Data: ciphertext, Fingerprint: crypto.Hash(test.canonical), IdentityFingerprint: legacyIdentity, SecretVersion: 1, AuthState: models.CredentialAuthStateReady, Status: models.CredentialStatusActive}
+			mustCreate(t, db, &row)
+			observation := models.CredentialObservation{CredentialID: row.ID, IdentityFingerprint: legacyIdentity, SchemaVersion: 1, ObservationVersion: 1, SnapshotJSON: models.JSON(`{}`), State: models.CredentialObservationFresh}
+			mustCreate(t, db, &observation)
+			channels, subscriptions := testSubscriptionRuntime(t)
+			driver, ok := subscriptions.Driver(test.channel)
+			if !ok {
+				t.Fatal("missing driver")
+			}
+			parsed, err := driver.Parse([]byte(test.canonical))
+			if err != nil {
+				t.Fatal(err)
+			}
+			if parsed.Identity() != "account/scope" {
+				t.Fatalf("derived identity = %q", parsed.Identity())
+			}
+			registry := state.NewCredentialRegistry()
+			if err := loader.NewWithCredentialValidation(db, state.NewManager(), registry, channels, subscriptions, crypto).Load(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			var after models.Credential
+			if err := db.First(&after, row.ID).Error; err != nil {
+				t.Fatal(err)
+			}
+			if after.Data != row.Data || after.Fingerprint != row.Fingerprint || after.IdentityFingerprint != legacyIdentity || after.SecretVersion != 1 {
+				t.Fatal("loading rewrote existing subscription data")
+			}
+			ref, ok := registry.CredentialRef(row.ID)
+			expectedGeneration := loader.CredentialIdentityGeneration(legacyIdentity, string(test.channel), "subscription", json.RawMessage(`{}`))
+			if !ok || ref.IdentityGeneration != expectedGeneration {
+				t.Fatal("loading changed the stored identity generation")
+			}
+			var savedObservation models.CredentialObservation
+			if err := db.First(&savedObservation, row.ID).Error; err != nil {
+				t.Fatal(err)
+			}
+			if savedObservation.IdentityFingerprint != after.IdentityFingerprint || savedObservation.State != models.CredentialObservationFresh {
+				t.Fatal("loading invalidated the existing observation")
+			}
+		})
+	}
+}
@@ -0,0 +1,79 @@
+package subscription
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"gpt-load/internal/channel"
+	"gpt-load/internal/storage/models"
+	"gpt-load/internal/subscription/providers/codex"
+	subscriptionruntime "gpt-load/internal/subscription/runtime"
+)
+
+func TestCredentialRefreshPreservesStoredIdentityWithOptionalUserClaims(t *testing.T) {
+	t.Parallel()
+	for _, test := range []struct {
+		name, oldUser, newAccount, newUser string
+		allowed                            bool
+	}{
+		{"same identity", "user-one", "account-one", "user-one", true},
+		{"legacy metadata still absent", "", "account-one", "", true},
+		{"enriched metadata", "", "account-one", "user-one", true},
+		{"missing metadata", "user-one", "account-one", "", false},
+		{"different user", "user-one", "account-one", "user-two", false},
+		{"different account", "user-one", "account-two", "user-one", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			makeCredential := func(account, user, token string, expires time.Time) []byte {
+				value := codex.Credential{Type: "codex", AccountID: account, AccessToken: "access-" + token, RefreshToken: "refresh-" + token, Expire: expires.UTC().Format(time.RFC3339)}
+				if user != "" {
+					claims, err := json.Marshal(map[string]any{"https://api.openai.com/auth": map[string]string{"chatgpt_account_id": account, "chatgpt_user_id": user}})
+					if err != nil {
+						t.Fatal(err)
+					}
+					value.IDToken = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".signature"
+				}
+				raw, err := codex.MarshalCredential(value)
+				if err != nil {
+					t.Fatal(err)
+				}
+				return raw
+			}
+			manager, db, registry, crypto, row := newCredentialManagerFixture(t, makeCredential("account-one", test.oldUser, "current", time.Now().Add(time.Minute)))
+			before, ok := registry.CredentialRef(row.ID)
+			if !ok {
+				t.Fatal("missing credential")
+			}
+			manager.refresh = func(_ context.Context, driver subscriptionruntime.Driver, _ subscriptionruntime.Credential) (subscriptionruntime.Credential, error) {
+				return driver.Parse(makeCredential(test.newAccount, test.newUser, "next", time.Now().Add(time.Hour)))
+			}
+			_, evidence := manager.Prepare(t.Context(), channel.Codex, credentialSnapshot(t, row, crypto), false)
+			if test.allowed && evidence != nil {
+				t.Fatalf("refresh rejected optional identity metadata: %s", evidence.Code)
+			}
+			if !test.allowed && (evidence == nil || evidence.Code != "refresh_identity_changed") {
+				t.Fatalf("changed identity evidence = %#v", evidence)
+			}
+			var stored models.Credential
+			if err := db.First(&stored, row.ID).Error; err != nil {
+				t.Fatal(err)
+			}
+			after, ok := registry.CredentialRef(row.ID)
+			if !ok {
+				t.Fatal("missing refreshed credential")
+			}
+			if stored.IdentityFingerprint != row.IdentityFingerprint || after.IdentityGeneration != before.IdentityGeneration {
+				t.Fatal("refresh rewrote durable identity or runtime generation")
+			}
+			if test.allowed && (stored.SecretVersion != 2 || stored.AuthState != models.CredentialAuthStateReady) {
+				t.Fatal("refreshed credential was not persisted")
+			}
+			if !test.allowed && (stored.Data != row.Data || stored.SecretVersion != 1 || stored.AuthState != models.CredentialAuthStateReauthorizationRequired) {
+				t.Fatal("changed identity overwrote the credential")
+			}
+		})
+	}
+}
@@ -300,7 +300,7 @@ func (manager *CredentialManager) refreshCredentialLocked(
 		}
 		return subscriptionruntime.Credential{}, authEvidence(evidenceCode)
 	}
-	if refreshed.Identity() == "" || refreshed.Identity() != current.Identity() {
+	if !subscriptionruntime.RefreshPreservesIdentity(driver, current, refreshed) {
 		if err := manager.transitionAuthState(ctx, row, row.SecretVersion, models.CredentialAuthStateReauthorizationRequired, "refresh_identity_changed"); err != nil {
 			manager.registry.SetCredentialAuthState(row.ID, state.CredentialAuthStateOutcomeUnknown)
 			return subscriptionruntime.Credential{}, localEvidence("refresh_state_commit_failed", "subscription credential state could not be saved")
@@ -58,6 +58,19 @@ func (*claudeDriver) Refresh(ctx context.Context, current subscriptionruntime.Cr
 	return claudeRuntimeCredential(refreshed, canonical), nil
 }

+func (*claudeDriver) MatchesRefreshIdentity(current, refreshed subscriptionruntime.Credential) bool {
+	before, err := ParseCredentialJSON(current.Canonical())
+	if err != nil {
+		return false
+	}
+	after, err := ParseCredentialJSON(refreshed.Canonical())
+	if err != nil || before.AccountUUID != after.AccountUUID {
+		return false
+	}
+	// 桥接会保留未返回的旧组织；已经确认的组织不能被清空或替换。
+	return before.OrganizationUUID == "" || before.OrganizationUUID == after.OrganizationUUID
+}
+
 func (*claudeDriver) ClassifyRefreshFailure(err error) subscriptionruntime.RefreshFailureDecision {
 	var tokenErr *TokenEndpointError
 	if errors.Is(err, ErrCredentialIdentityChanged) ||
@@ -214,7 +227,7 @@ func claudeRuntimeCredential(value Credential, canonical []byte) subscriptionrun
 	}
 	return subscriptionruntime.NewCredential(
 		canonical,
-		strings.TrimSpace(value.AccountUUID),
+		credentialIdentity(value),
 		account,
 		expiresAt,
 		expires,
@@ -223,3 +236,11 @@ func claudeRuntimeCredential(value Credential, canonical []byte) subscriptionrun
 }

 var _ subscriptionruntime.BrowserAuthorizationDriver = (*claudeDriver)(nil)
+
+func credentialIdentity(value Credential) string {
+	accountID := strings.TrimSpace(value.AccountUUID)
+	if organizationID := strings.TrimSpace(value.OrganizationUUID); organizationID != "" {
+		return accountID + "/" + organizationID
+	}
+	return accountID
+}
@@ -26,7 +26,7 @@ func TestClaudeDriverProducesProviderNeutralCredential(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if credential.Identity() != "account-one" || credential.Account().Email != "owner@example.com" ||
+	if credential.Identity() != "account-one/org-one" || credential.Account().Email != "owner@example.com" ||
 		!credential.Account().ExpiresAtKnown {
 		t.Fatalf("credential metadata = %q %#v", credential.Identity(), credential.Account())
 	}
@@ -44,7 +44,11 @@ func ParseCredentialJSON(raw []byte) (Credential, error) {
 	if err != nil {
 		return Credential{}, err
 	}
-	return credentialFromBridge(parsed), nil
+	value := credentialFromBridge(parsed)
+	if err := validateCredentialIdentity(value); err != nil {
+		return Credential{}, err
+	}
+	return value, nil
 }

 // MarshalCredential returns the canonical persisted representation.
@@ -266,7 +266,7 @@ func codexRuntimeCredential(value Credential, canonical []byte) subscriptionrunt
 	if refreshed, err := time.Parse(time.RFC3339, strings.TrimSpace(value.LastRefresh)); err == nil {
 		account.LastRefresh, account.LastRefreshKnown = refreshed, true
 	}
-	return subscriptionruntime.NewCredential(canonical, strings.TrimSpace(value.AccountID), account, expiresAt, expires, value.SecretValues())
+	return subscriptionruntime.NewCredential(canonical, credentialIdentity(value), account, expiresAt, expires, value.SecretValues())
 }

 var _ subscriptionruntime.BrowserAuthorizationDriver = (*codexDriver)(nil)
@@ -0,0 +1,79 @@
+package codex
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"strings"
+
+	subscriptionruntime "gpt-load/internal/subscription/runtime"
+)
+
+func (*codexDriver) MatchesRefreshIdentity(current, refreshed subscriptionruntime.Credential) bool {
+	before, err := ParseCredentialJSON(current.Canonical())
+	if err != nil {
+		return false
+	}
+	after, err := ParseCredentialJSON(refreshed.Canonical())
+	if err != nil || before.AccountID != after.AccountID {
+		return false
+	}
+	beforeUser, afterUser := credentialUserID(before), credentialUserID(after)
+	// 旧凭据允许补全身份，但不能把已经确认的用户降级为未知用户。
+	return beforeUser == "" || beforeUser == afterUser
+}
+
+// 身份只从既有令牌派生，不能改变参与持久化内容指纹计算的 canonical JSON。
+func credentialIdentity(value Credential) string {
+	accountID := strings.TrimSpace(value.AccountID)
+	if userID := credentialUserID(value); userID != "" {
+		return accountID + "/" + userID
+	}
+	return accountID
+}
+
+func credentialUserID(value Credential) string {
+	for _, token := range []string{value.IDToken, value.AccessToken} {
+		if userID := tokenUserID(token); userID != "" {
+			return userID
+		}
+	}
+	return ""
+}
+
+// 不能让旧 ID token 掩盖实际用于请求的新 access token 所属用户。
+func validateCredentialIdentity(value Credential) error {
+	idUser, accessUser := tokenUserID(value.IDToken), tokenUserID(value.AccessToken)
+	if idUser != "" && accessUser != "" && idUser != accessUser {
+		return ErrCredentialIdentityChanged
+	}
+	return nil
+}
+
+// JWT 仅用于读取已持有凭据的身份元数据，不作为签名或登录认证。
+func tokenUserID(token string) string {
+	parts := strings.Split(strings.TrimSpace(token), ".")
+	if len(parts) != 3 {
+		return ""
+	}
+	payload, err := base64.RawURLEncoding.DecodeString(parts[1])
+	if err != nil {
+		payload, err = base64.URLEncoding.DecodeString(parts[1])
+		if err != nil {
+			return ""
+		}
+	}
+	defer clear(payload)
+	var claims struct {
+		Auth struct {
+			ChatGPTUserID string `json:"chatgpt_user_id"`
+			UserID        string `json:"user_id"`
+		} `json:"https://api.openai.com/auth"`
+	}
+	if json.Unmarshal(payload, &claims) != nil {
+		return ""
+	}
+	if userID := strings.TrimSpace(claims.Auth.ChatGPTUserID); userID != "" {
+		return userID
+	}
+	return strings.TrimSpace(claims.Auth.UserID)
+}
@@ -0,0 +1,131 @@
+package codex
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	subscriptionruntime "gpt-load/internal/subscription/runtime"
+)
+
+func identityTestToken(t *testing.T, claims map[string]string, padded bool) string {
+	t.Helper()
+	raw, err := json.Marshal(map[string]any{"https://api.openai.com/auth": claims})
+	if err != nil {
+		t.Fatal(err)
+	}
+	encoding := base64.RawURLEncoding
+	if padded {
+		encoding = base64.URLEncoding
+	}
+	return "e30." + encoding.EncodeToString(raw) + ".signature"
+}
+
+func TestCodexIdentityUsesUserClaimsWithoutChangingCanonical(t *testing.T) {
+	t.Parallel()
+	user := identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one"}, false)
+	alias := identityTestToken(t, map[string]string{"user_id": "user-two"}, true)
+	for _, test := range []struct{ name, idToken, accessToken, want string }{
+		{"id token", user, "access", "workspace/user-one"},
+		{"access token", "not-a-jwt", user, "workspace/user-one"},
+		{"user id alias", alias, "access", "workspace/user-two"},
+		{"access token alias", "", alias, "workspace/user-two"},
+		{"claim precedence", identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one", "user_id": "user-two"}, false), "access", "workspace/user-one"},
+		{"blank primary claim", identityTestToken(t, map[string]string{"chatgpt_user_id": " ", "user_id": "user-two"}, false), "access", "workspace/user-two"},
+		{"missing claims", identityTestToken(t, map[string]string{}, false), "access", "workspace"},
+		{"invalid token", "e30.!.signature", "access", "workspace"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			// 固定旧格式的字段顺序，身份推导不得改变 canonical 字节或持久化派生字段。
+			raw := []byte(fmt.Sprintf(`{"type":"codex"%s,"access_token":%q,"refresh_token":"refresh","account_id":"workspace","email":"owner@example.com"}`, identityTestIDTokenField(test.idToken), test.accessToken))
+			credential, err := newCodexDriver().Parse(raw)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(credential.Canonical()) != string(raw) {
+				t.Fatal("identity derivation changed canonical credential bytes")
+			}
+			if credential.Identity() != test.want {
+				t.Fatalf("identity = %q, want %q", credential.Identity(), test.want)
+			}
+		})
+	}
+}
+
+func identityTestIDTokenField(token string) string {
+	if token == "" {
+		return ""
+	}
+	return fmt.Sprintf(`,"id_token":%q`, token)
+}
+
+func TestCodexIdentityRejectsConflictingTokenUsers(t *testing.T) {
+	t.Parallel()
+	for _, accessClaim := range []string{"chatgpt_user_id", "user_id"} {
+		t.Run(accessClaim, func(t *testing.T) {
+			value := Credential{Type: "codex", AccountID: "workspace", RefreshToken: "refresh", IDToken: identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one"}, false), AccessToken: identityTestToken(t, map[string]string{accessClaim: "user-two"}, false)}
+			raw, err := json.Marshal(value)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if _, err := newCodexDriver().Parse(raw); err == nil {
+				t.Fatal("conflicting token identities were accepted")
+			}
+			if _, err := MarshalCredential(value); err == nil {
+				t.Fatal("conflicting token identities were canonicalized")
+			}
+		})
+	}
+}
+
+func TestCodexRefreshWithoutNewIDTokenPreservesKnownIdentity(t *testing.T) {
+	previous := http.DefaultTransport
+	t.Cleanup(func() { http.DefaultTransport = previous })
+	http.DefaultTransport = targetRoundTripper(func(r *http.Request) (*http.Response, error) {
+		if r.Method != http.MethodPost || r.URL.String() != "https://auth.openai.com/oauth/token" {
+			t.Errorf("unexpected refresh request: %s %s", r.Method, r.URL)
+		}
+		return &http.Response{
+			StatusCode: http.StatusOK, Header: make(http.Header), Request: r,
+			Body: io.NopCloser(strings.NewReader(`{"access_token":"new-access","refresh_token":"new-refresh","expires_in":3600}`)),
+		}, nil
+	})
+	userToken := identityTestToken(t, map[string]string{"chatgpt_account_id": "workspace", "chatgpt_user_id": "user-one"}, false)
+	for _, test := range []struct {
+		name, idToken, accessToken string
+		allowed                    bool
+	}{
+		{"retained ID token", userToken, "old-access", true},
+		{"last user claim only in replaced access token", "", userToken, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			raw, err := MarshalCredential(Credential{Type: "codex", AccountID: "workspace", IDToken: test.idToken, AccessToken: test.accessToken, RefreshToken: "old-refresh"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			driver := newCodexDriver()
+			current, err := driver.Parse(raw)
+			if err != nil {
+				t.Fatal(err)
+			}
+			refreshed, err := driver.Refresh(t.Context(), current)
+			if err != nil {
+				t.Fatal(err)
+			}
+			value, err := ParseCredentialJSON(refreshed.Canonical())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if value.IDToken != test.idToken || value.AccessToken != "new-access" || value.RefreshToken != "new-refresh" {
+				t.Fatal("refresh did not retain the old ID token while replacing access and refresh tokens")
+			}
+			if allowed := subscriptionruntime.RefreshPreservesIdentity(driver, current, refreshed); allowed != test.allowed {
+				t.Fatalf("refresh allowed = %v, want %v", allowed, test.allowed)
+			}
+		})
+	}
+}
@@ -57,7 +57,7 @@ func NewCredential(canonical []byte, identity string, account Account, expiresAt
 // Canonical returns an independent copy suitable for encryption or adapter decoding.
 func (credential Credential) Canonical() []byte { return append([]byte(nil), credential.canonical...) }

-// Identity returns the stable provider account identity used across refreshes.
+// Identity returns the provider account identity used for duplicate checks.
 func (credential Credential) Identity() string { return credential.identity }

 // Account returns safe display metadata.
@@ -157,6 +157,22 @@ type Driver interface {
 	ClassifyRefreshFailure(error) RefreshFailureDecision
 }

+// RefreshIdentityMatcher 由身份包含可选字段的渠道实现；允许补全，但禁止已知身份丢失或变化。
+type RefreshIdentityMatcher interface {
+	MatchesRefreshIdentity(current, refreshed Credential) bool
+}
+
+// RefreshPreservesIdentity 默认要求身份完全相同；可选字段的规则由渠道负责。
+func RefreshPreservesIdentity(driver Driver, current, refreshed Credential) bool {
+	if current.Identity() == "" || refreshed.Identity() == "" {
+		return false
+	}
+	if matcher, ok := driver.(RefreshIdentityMatcher); ok {
+		return matcher.MatchesRefreshIdentity(current, refreshed)
+	}
+	return current.Identity() == refreshed.Identity()
+}
+
 // CredentialFileImporter is an optional, narrow preprocessing capability for
 // OAuth file formats that cannot safely provide a canonical credential without
 // one provider-bound verification step. Existing drivers continue to use Parse.