From 6793d5ac85f4ca469cc49a9cec29285c78e445b1 Mon Sep 17 00:00:00 2001 From: Jarl Date: Mon, 17 Aug 2026 19:38:03 +0800 Subject: [PATCH] fix(openai): make Codex convergence identity consistent --- backend/internal/repository/account_repo.go | 43 ++- ...ccount_repo_codex_fingerprint_seed_test.go | 123 +++++++ ...erprint_seed_migration_integration_test.go | 150 ++++++++ .../internal/repository/scheduler_cache.go | 2 + .../repository/scheduler_cache_unit_test.go | 4 + ...t_codex_fingerprint_seed_lifecycle_test.go | 245 +++++++++++++ backend/internal/service/account_service.go | 9 +- ...ccount_test_service_openai_compact_test.go | 11 +- backend/internal/service/admin_account.go | 48 ++- .../service/openai_codex_fingerprint.go | 336 ++++++++++++++---- .../service/openai_codex_fingerprint_test.go | 325 +++++++++++++++-- .../service/openai_gateway_forward.go | 5 +- .../service/openai_gateway_passthrough.go | 1 + .../service/openai_oauth_passthrough_test.go | 206 +++++++++++ .../service/openai_ws_forwarder_payload.go | 9 +- .../openai_ws_forwarder_success_test.go | 137 +++++++ .../service/openai_ws_forwarder_v2.go | 15 +- backend/internal/service/openai_ws_pool.go | 100 ++++-- .../internal/service/openai_ws_pool_test.go | 200 +++++++++++ .../225_backfill_codex_fingerprint_seed.sql | 21 ++ 20 files changed, 1832 insertions(+), 158 deletions(-) create mode 100644 backend/internal/repository/account_repo_codex_fingerprint_seed_test.go create mode 100644 backend/internal/repository/codex_fingerprint_seed_migration_integration_test.go create mode 100644 backend/internal/service/account_codex_fingerprint_seed_lifecycle_test.go create mode 100644 backend/migrations/225_backfill_codex_fingerprint_seed.sql diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 010442eb6b..664a845460 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -71,6 +71,39 @@ var schedulerNeutralExtraKeys = map[string]struct{}{ const postgresParameterBatchSize = 50000 +const codexFingerprintSeedCanonicalPattern = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +const codexFingerprintNilSeed = "00000000-0000-0000-0000-000000000000" + +func codexFingerprintSeedValidSQL(extraExpr string) string { + value := "(" + extraExpr + " ->> 'codex_fingerprint_seed')" + return "(" + value + " ~ '" + codexFingerprintSeedCanonicalPattern + "' AND " + value + " <> '" + codexFingerprintNilSeed + "')" +} + +func ensureCodexFingerprintSeedSQL(extraExpr string) string { + return "CASE WHEN platform = 'openai' AND type = 'oauth' THEN " + + "jsonb_set(" + extraExpr + ", '{codex_fingerprint_seed}', " + + "CASE WHEN " + codexFingerprintSeedValidSQL("extra") + + " THEN to_jsonb(extra ->> 'codex_fingerprint_seed') ELSE to_jsonb(gen_random_uuid()::text) END, true) " + + "ELSE " + extraExpr + " END" +} + +func stripCodexFingerprintSeedFromExtraUpdate(extra map[string]any) map[string]any { + if extra == nil { + return nil + } + if _, exists := extra["codex_fingerprint_seed"]; !exists { + return extra + } + stripped := make(map[string]any, len(extra)-1) + for key, value := range extra { + if key == "codex_fingerprint_seed" { + continue + } + stripped[key] = value + } + return stripped +} + // NewAccountRepository 创建账户仓储实例。 // 这是对外暴露的构造函数，返回接口类型以便于依赖注入。 func NewAccountRepository(client *dbent.Client, sqlDB *sql.DB, schedulerCache service.SchedulerCache) service.AccountRepository { @@ -2520,6 +2553,7 @@ func (r *accountRepository) AutoPauseExpiredAccounts(ctx context.Context, now ti } func (r *accountRepository) UpdateExtra(ctx context.Context, id int64, updates map[string]any) error { + updates = stripCodexFingerprintSeedFromExtraUpdate(updates) if len(updates) == 0 { return nil } @@ -2552,6 +2586,9 @@ func (r *accountRepository) UpdateExtra(ctx context.Context, id int64, updates m if clearProbeSnapshot { extraExpression = "(" + extraExpression + ") - 'upstream_billing_probe'" } + if service.ShouldEnsureCodexFingerprintSeedForExtraUpdates(updates) { + extraExpression = ensureCodexFingerprintSeedSQL(extraExpression) + } result, err := client.ExecContext( ctx, "UPDATE accounts SET extra = "+extraExpression+", updated_at = NOW() WHERE id = $2 AND deleted_at IS NULL", @@ -2793,6 +2830,7 @@ func (r *accountRepository) BulkUpdate(ctx context.Context, ids []int64, updates if len(ids) == 0 { return 0, nil } + updates.Extra = stripCodexFingerprintSeedFromExtraUpdate(updates.Extra) setClauses := make([]string, 0, 8) args := make([]any, 0, 8) @@ -2880,7 +2918,7 @@ func (r *accountRepository) BulkUpdate(ctx context.Context, ids []int64, updates " AND "+ollamaCloudBaseURLMatchesSQL(credentialPlaceholder+"::jsonb ->> 'base_url'")+")") } - if len(updates.Extra) > 0 || len(ollamaGroupIdentityChanges) > 0 || ollamaProxyIdentityChanged != "" { + if len(updates.Extra) > 0 || len(ollamaGroupIdentityChanges) > 0 || ollamaProxyIdentityChanged != "" || updates.EnsureCodexFingerprintSeed { extraExpression := "COALESCE(extra, '{}'::jsonb)" if len(updates.Extra) > 0 { payload, err := json.Marshal(updates.Extra) @@ -2919,6 +2957,9 @@ func (r *accountRepository) BulkUpdate(ctx context.Context, ids []int64, updates } else if snapshotIdentityChanged != "" { extraExpression = "CASE WHEN " + snapshotIdentityChanged + " THEN (" + extraExpression + ") - 'ollama_cloud_usage_snapshot' ELSE " + extraExpression + " END" } + if updates.EnsureCodexFingerprintSeed { + extraExpression = ensureCodexFingerprintSeedSQL(extraExpression) + } setClauses = append(setClauses, "extra = "+extraExpression) } diff --git a/backend/internal/repository/account_repo_codex_fingerprint_seed_test.go b/backend/internal/repository/account_repo_codex_fingerprint_seed_test.go new file mode 100644 index 0000000000..1ad1723748 --- /dev/null +++ b/backend/internal/repository/account_repo_codex_fingerprint_seed_test.go @@ -0,0 +1,123 @@ +package repository + +import ( + "context" + "errors" + "regexp" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + dbent "github.com/Wei-Shaw/sub2api/ent" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" + + "entgo.io/ent/dialect" + entsql "entgo.io/ent/dialect/sql" +) + +func TestBulkUpdateEnsuresCodexFingerprintSeedWithPerRowSQL(t *testing.T) { + exec := &recordingSQLExecutor{result: rowsAffectedResult(0)} + repo := newAccountRepositoryWithSQL(nil, exec, nil) + + _, err := repo.BulkUpdate(context.Background(), []int64{27, 28}, service.AccountBulkUpdate{ + Extra: map[string]any{ + "codex_fingerprint_mode": "session", + "codex_fingerprint_seed": "22222222-2222-4222-8222-222222222222", + }, + EnsureCodexFingerprintSeed: true, + }) + + require.NoError(t, err) + require.NotEmpty(t, exec.execQueries) + query := normalizeSQLWhitespace(exec.execQueries[0]) + require.Contains(t, query, "jsonb_set") + require.Contains(t, query, "gen_random_uuid()::text") + require.Contains(t, query, "platform = 'openai' AND type = 'oauth'") + require.Contains(t, query, "to_jsonb(extra ->> 'codex_fingerprint_seed')") + require.Contains(t, query, codexFingerprintSeedCanonicalPattern) + require.NotContains(t, query, "22222222-2222-4222-8222-222222222222") + require.NotEmpty(t, exec.execArgs) + payload, ok := exec.execArgs[0][0].([]byte) + require.True(t, ok) + require.Equal(t, `{"codex_fingerprint_mode":"session"}`, string(payload)) +} + +func TestUpdateExtraEnsuresCodexFingerprintSeedAtomicallyWhenEnabling(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, db))) + t.Cleanup(func() { _ = client.Close() }) + + mock.ExpectBegin() + mock.ExpectExec(`(?s)UPDATE accounts SET extra = .*jsonb_set.*gen_random_uuid\(\)::text.*WHERE id = \$2 AND deleted_at IS NULL`). + WithArgs(`{"codex_fingerprint_mode":"device"}`, int64(27)). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(regexp.QuoteMeta("INSERT INTO scheduler_outbox")). + WithArgs(service.SchedulerOutboxEventAccountChanged, int64(27), nil, nil, sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectCommit() + repo := newAccountRepositoryWithSQL(client, db, nil) + + err = repo.UpdateExtra(context.Background(), 27, map[string]any{ + "codex_fingerprint_mode": "device", + "codex_fingerprint_seed": "22222222-2222-4222-8222-222222222222", + }) + + require.NoError(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestBulkUpdateCodexFingerprintSeedRollsBackWhenUpdateFails(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, db))) + t.Cleanup(func() { _ = client.Close() }) + + mock.ExpectBegin() + mock.ExpectExec(`(?s)UPDATE accounts SET extra = .*gen_random_uuid\(\)::text.*WHERE id = ANY\(\$2\)`). + WithArgs(sqlmock.AnyArg(), `{27,28}`). + WillReturnError(errors.New("update failed")) + mock.ExpectRollback() + + repo := newAccountRepositoryWithSQL(client, db, nil) + rows, err := repo.BulkUpdate(context.Background(), []int64{27, 28}, service.AccountBulkUpdate{ + Extra: map[string]any{ + "codex_fingerprint_mode": "session", + }, + EnsureCodexFingerprintSeed: true, + }) + + require.EqualError(t, err, "update failed") + require.Zero(t, rows) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestBulkUpdateCodexFingerprintSeedRollsBackWhenOutboxFails(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, db))) + t.Cleanup(func() { _ = client.Close() }) + + mock.ExpectBegin() + mock.ExpectExec(`(?s)UPDATE accounts SET extra = .*gen_random_uuid\(\)::text.*WHERE id = ANY\(\$2\)`). + WithArgs(sqlmock.AnyArg(), `{27,28}`). + WillReturnResult(sqlmock.NewResult(0, 2)) + mock.ExpectExec(regexp.QuoteMeta("INSERT INTO scheduler_outbox")). + WillReturnError(errors.New("outbox failed")) + mock.ExpectRollback() + + repo := newAccountRepositoryWithSQL(client, db, nil) + rows, err := repo.BulkUpdate(context.Background(), []int64{27, 28}, service.AccountBulkUpdate{ + Extra: map[string]any{ + "codex_fingerprint_mode": "full", + }, + EnsureCodexFingerprintSeed: true, + }) + + require.EqualError(t, err, "outbox failed") + require.Zero(t, rows) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/backend/internal/repository/codex_fingerprint_seed_migration_integration_test.go b/backend/internal/repository/codex_fingerprint_seed_migration_integration_test.go new file mode 100644 index 0000000000..70ae61eb8e --- /dev/null +++ b/backend/internal/repository/codex_fingerprint_seed_migration_integration_test.go @@ -0,0 +1,150 @@ +//go:build integration + +package repository + +import ( + "context" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + dbmigrations "github.com/Wei-Shaw/sub2api/migrations" + "github.com/google/uuid" + "github.com/lib/pq" + "github.com/stretchr/testify/require" +) + +func requireCanonicalUUIDString(t *testing.T, value string) { + t.Helper() + parsed, err := uuid.Parse(value) + require.NoError(t, err) + require.NotEqual(t, uuid.Nil, parsed) + require.Equal(t, parsed.String(), value) +} + +func TestMigration225BackfillsOnlyEnabledOpenAIOAuthMissingOrMalformedSeeds(t *testing.T) { + tx := testTx(t) + ctx := context.Background() + migrationSQL, err := dbmigrations.FS.ReadFile("225_backfill_codex_fingerprint_seed.sql") + require.NoError(t, err) + + var missingID, blankID, malformedID, validID, offID, apiKeyID int64 + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-missing', 'openai', 'oauth', '{"codex_fingerprint_mode":"session"}'::jsonb) +RETURNING id +`).Scan(&missingID)) + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-blank', 'openai', 'oauth', '{"codex_fingerprint_mode":"device","codex_fingerprint_seed":""}'::jsonb) +RETURNING id +`).Scan(&blankID)) + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-malformed', 'openai', 'oauth', '{"codex_fingerprint_mode":"full","codex_fingerprint_seed":"BAD"}'::jsonb) +RETURNING id +`).Scan(&malformedID)) + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-valid', 'openai', 'oauth', '{"codex_fingerprint_mode":"session","codex_fingerprint_seed":"11111111-1111-4111-8111-111111111111"}'::jsonb) +RETURNING id +`).Scan(&validID)) + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-off', 'openai', 'oauth', '{"codex_fingerprint_mode":"off"}'::jsonb) +RETURNING id +`).Scan(&offID)) + require.NoError(t, tx.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ('migration-225-apikey', 'openai', 'apikey', '{"codex_fingerprint_mode":"session"}'::jsonb) +RETURNING id +`).Scan(&apiKeyID)) + + _, err = tx.ExecContext(ctx, string(migrationSQL)) + require.NoError(t, err) + + seedsAfterFirst := map[int64]string{} + for _, id := range []int64{missingID, blankID, malformedID, validID} { + var seed string + require.NoError(t, tx.QueryRowContext(ctx, `SELECT extra->>'codex_fingerprint_seed' FROM accounts WHERE id = $1`, id).Scan(&seed)) + requireCanonicalUUIDString(t, seed) + seedsAfterFirst[id] = seed + } + require.Equal(t, "11111111-1111-4111-8111-111111111111", seedsAfterFirst[validID]) + + for _, id := range []int64{offID, apiKeyID} { + var hasSeed bool + require.NoError(t, tx.QueryRowContext(ctx, `SELECT extra ? 'codex_fingerprint_seed' FROM accounts WHERE id = $1`, id).Scan(&hasSeed)) + require.False(t, hasSeed) + } + + _, err = tx.ExecContext(ctx, string(migrationSQL)) + require.NoError(t, err) + + for id, want := range seedsAfterFirst { + var got string + require.NoError(t, tx.QueryRowContext(ctx, `SELECT extra->>'codex_fingerprint_seed' FROM accounts WHERE id = $1`, id).Scan(&got)) + require.Equal(t, want, got) + } +} + +func TestBulkUpdateGeneratesDistinctStableCodexFingerprintSeedsPerEligibleRow(t *testing.T) { + ctx := context.Background() + testName := "bulk-codex-seed-" + uuid.NewString() + type fixture struct { + name string + accountType string + extra string + } + fixtures := []fixture{ + {name: testName + "-missing-a", accountType: service.AccountTypeOAuth, extra: `{}`}, + {name: testName + "-missing-b", accountType: service.AccountTypeOAuth, extra: `{"codex_fingerprint_seed":"BAD"}`}, + {name: testName + "-valid", accountType: service.AccountTypeOAuth, extra: `{"codex_fingerprint_seed":"11111111-1111-4111-8111-111111111111"}`}, + {name: testName + "-apikey", accountType: service.AccountTypeAPIKey, extra: `{}`}, + } + + ids := make([]int64, 0, len(fixtures)) + for _, f := range fixtures { + var id int64 + require.NoError(t, integrationDB.QueryRowContext(ctx, ` +INSERT INTO accounts (name, platform, type, extra) +VALUES ($1, 'openai', $2, $3::jsonb) +RETURNING id +`, f.name, f.accountType, f.extra).Scan(&id)) + ids = append(ids, id) + } + t.Cleanup(func() { + _, _ = integrationDB.ExecContext(context.Background(), `DELETE FROM scheduler_outbox WHERE account_id = ANY($1)`, pq.Array(ids)) + _, _ = integrationDB.ExecContext(context.Background(), `DELETE FROM accounts WHERE id = ANY($1)`, pq.Array(ids)) + }) + + repo := newAccountRepositoryWithSQL(testEntClient(t), integrationDB, nil) + updates := service.AccountBulkUpdate{ + Extra: map[string]any{ + "codex_fingerprint_mode": "session", + }, + EnsureCodexFingerprintSeed: true, + } + rows, err := repo.BulkUpdate(ctx, ids, updates) + require.NoError(t, err) + require.Equal(t, int64(len(ids)), rows) + + readSeed := func(id int64) string { + t.Helper() + var seed string + require.NoError(t, integrationDB.QueryRowContext(ctx, `SELECT COALESCE(extra->>'codex_fingerprint_seed', '') FROM accounts WHERE id = $1`, id).Scan(&seed)) + return seed + } + firstSeeds := []string{readSeed(ids[0]), readSeed(ids[1]), readSeed(ids[2]), readSeed(ids[3])} + requireCanonicalUUIDString(t, firstSeeds[0]) + requireCanonicalUUIDString(t, firstSeeds[1]) + require.NotEqual(t, firstSeeds[0], firstSeeds[1], "gen_random_uuid must be evaluated per eligible row") + require.Equal(t, "11111111-1111-4111-8111-111111111111", firstSeeds[2]) + require.Empty(t, firstSeeds[3], "API-key accounts must not receive a Codex fingerprint seed") + + rows, err = repo.BulkUpdate(ctx, ids, updates) + require.NoError(t, err) + require.Equal(t, int64(len(ids)), rows) + for i, want := range firstSeeds { + require.Equal(t, want, readSeed(ids[i]), "retry must not rotate an existing valid seed") + } +} diff --git a/backend/internal/repository/scheduler_cache.go b/backend/internal/repository/scheduler_cache.go index b03e4e6637..b53c51d43f 100644 --- a/backend/internal/repository/scheduler_cache.go +++ b/backend/internal/repository/scheduler_cache.go @@ -1000,6 +1000,8 @@ func filterSchedulerExtra(extra map[string]any) map[string]any { "openai_ws_force_http", "openai_responses_mode", "openai_responses_supported", + "codex_fingerprint_mode", + "codex_fingerprint_seed", "codex_5h_used_percent", "codex_7d_used_percent", "codex_5h_reset_at", diff --git a/backend/internal/repository/scheduler_cache_unit_test.go b/backend/internal/repository/scheduler_cache_unit_test.go index d0373f16c7..354cf6d8aa 100644 --- a/backend/internal/repository/scheduler_cache_unit_test.go +++ b/backend/internal/repository/scheduler_cache_unit_test.go @@ -309,6 +309,8 @@ func TestBuildSchedulerMetadataAccount_KeepsOpenAIWSFlags(t *testing.T) { "openai_ws_force_http": true, "openai_responses_mode": "force_chat_completions", "openai_responses_supported": false, + "codex_fingerprint_mode": "session", + "codex_fingerprint_seed": "11111111-1111-4111-8111-111111111111", "mixed_scheduling": true, "unused_large_field": "drop-me", }, @@ -321,6 +323,8 @@ func TestBuildSchedulerMetadataAccount_KeepsOpenAIWSFlags(t *testing.T) { require.Equal(t, true, got.Extra["openai_ws_force_http"]) require.Equal(t, "force_chat_completions", got.Extra["openai_responses_mode"]) require.Equal(t, false, got.Extra["openai_responses_supported"]) + require.Equal(t, "session", got.Extra["codex_fingerprint_mode"]) + require.Equal(t, "11111111-1111-4111-8111-111111111111", got.Extra["codex_fingerprint_seed"]) require.Equal(t, true, got.Extra["mixed_scheduling"]) require.Nil(t, got.Extra["unused_large_field"]) } diff --git a/backend/internal/service/account_codex_fingerprint_seed_lifecycle_test.go b/backend/internal/service/account_codex_fingerprint_seed_lifecycle_test.go new file mode 100644 index 0000000000..029b4fb009 --- /dev/null +++ b/backend/internal/service/account_codex_fingerprint_seed_lifecycle_test.go @@ -0,0 +1,245 @@ +package service + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +const userSuppliedCodexFingerprintSeed = "22222222-2222-4222-8222-222222222222" + +func requireValidCodexFingerprintSeed(t *testing.T, extra map[string]any) string { + t.Helper() + seed, ok := codexFingerprintSeed(extra) + require.True(t, ok, "expected valid canonical Codex fingerprint seed") + return seed +} + +func TestAdminCreateAccountStripsUserSeedAndCreatesFreshSeedWhenEnabled(t *testing.T) { + repo := &upstreamBillingProbeAccountRepo{} + svc := &adminServiceImpl{accountRepo: repo} + + created, err := svc.CreateAccount(context.Background(), &CreateAccountInput{ + Name: "codex-oauth", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + SkipDefaultGroupBind: true, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: userSuppliedCodexFingerprintSeed, + }, + }) + + require.NoError(t, err) + seed := requireValidCodexFingerprintSeed(t, created.Extra) + require.NotEqual(t, userSuppliedCodexFingerprintSeed, seed) + require.Equal(t, "session", created.Extra[codexFingerprintModeExtraKey]) +} + +func TestAdminUpdateAccountPreservesExistingSeedAndStripsUserSeed(t *testing.T) { + accountID := int64(201) + repo := &upstreamBillingProbeAccountRepo{accounts: map[int64]*Account{ + accountID: { + ID: accountID, + Name: "before", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: testCodexFingerprintSeed, + }, + }, + }} + svc := &adminServiceImpl{accountRepo: repo} + + updated, err := svc.UpdateAccount(context.Background(), accountID, &UpdateAccountInput{ + Extra: map[string]any{ + codexFingerprintModeExtraKey: "full", + codexFingerprintSeedExtraKey: userSuppliedCodexFingerprintSeed, + "custom": "value", + }, + }) + + require.NoError(t, err) + require.Equal(t, testCodexFingerprintSeed, requireValidCodexFingerprintSeed(t, updated.Extra)) + require.Equal(t, "full", updated.Extra[codexFingerprintModeExtraKey]) + require.Equal(t, "value", updated.Extra["custom"]) +} + +func TestAdminUpdateAccountInitializesSeedWhenFullEditEnables(t *testing.T) { + accountID := int64(202) + repo := &upstreamBillingProbeAccountRepo{accounts: map[int64]*Account{ + accountID: { + ID: accountID, + Name: "before", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "off", + codexFingerprintSeedExtraKey: "not-a-seed", + }, + }, + }} + + updated, err := (&adminServiceImpl{accountRepo: repo}).UpdateAccount(context.Background(), accountID, &UpdateAccountInput{ + Extra: map[string]any{codexFingerprintModeExtraKey: "device"}, + }) + + require.NoError(t, err) + require.NotEqual(t, "not-a-seed", requireValidCodexFingerprintSeed(t, updated.Extra)) + require.Equal(t, "device", updated.Extra[codexFingerprintModeExtraKey]) +} + +func TestAdminUpdateAccountDisableReenablePreservesValidSeed(t *testing.T) { + accountID := int64(203) + repo := &upstreamBillingProbeAccountRepo{accounts: map[int64]*Account{ + accountID: { + ID: accountID, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: testCodexFingerprintSeed, + }, + }, + }} + svc := &adminServiceImpl{accountRepo: repo} + + disabled, err := svc.UpdateAccount(context.Background(), accountID, &UpdateAccountInput{ + Extra: map[string]any{codexFingerprintModeExtraKey: "off"}, + }) + require.NoError(t, err) + require.Equal(t, testCodexFingerprintSeed, requireValidCodexFingerprintSeed(t, disabled.Extra)) + + reenabled, err := svc.UpdateAccount(context.Background(), accountID, &UpdateAccountInput{ + Extra: map[string]any{codexFingerprintModeExtraKey: "session"}, + }) + require.NoError(t, err) + require.Equal(t, testCodexFingerprintSeed, requireValidCodexFingerprintSeed(t, reenabled.Extra)) +} + +func TestAdminUpdateAccountExtraStripsSeedAndLeavesAtomicEnsureToRepository(t *testing.T) { + accountID := int64(204) + repo := &upstreamBillingProbeAccountRepo{accounts: map[int64]*Account{ + accountID: { + ID: accountID, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Extra: map[string]any{}, + }, + }} + + err := (&adminServiceImpl{accountRepo: repo}).UpdateAccountExtra(context.Background(), accountID, map[string]any{ + codexFingerprintModeExtraKey: "device", + codexFingerprintSeedExtraKey: userSuppliedCodexFingerprintSeed, + }) + + require.NoError(t, err) + require.Len(t, repo.updates[accountID], 1) + require.Equal(t, "device", repo.updates[accountID][0][codexFingerprintModeExtraKey]) + require.NotContains(t, repo.updates[accountID][0], codexFingerprintSeedExtraKey) +} + +func TestBulkUpdateAccountsDoesNotPrewriteCodexSeed(t *testing.T) { + repo := &upstreamBillingProbeAccountRepo{} + + result, err := (&adminServiceImpl{accountRepo: repo}).BulkUpdateAccounts(context.Background(), &BulkUpdateAccountsInput{ + AccountIDs: []int64{301, 302}, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: userSuppliedCodexFingerprintSeed, + }, + }) + + require.NoError(t, err) + require.Equal(t, 2, result.Success) + require.Empty(t, repo.updates, "bulk enable must not loop through UpdateExtra before BulkUpdate") + require.Len(t, repo.bulkUpdates, 1) + require.True(t, repo.bulkUpdates[0].EnsureCodexFingerprintSeed) + require.Equal(t, "session", repo.bulkUpdates[0].Extra[codexFingerprintModeExtraKey]) + require.NotContains(t, repo.bulkUpdates[0].Extra, codexFingerprintSeedExtraKey) +} + +type codexSeedDuplicateRepo struct { + *upstreamBillingProbeAccountRepo +} + +func (r *codexSeedDuplicateRepo) CreateWithAccountGroups(ctx context.Context, account *Account, _ []AccountGroup) error { + return r.Create(ctx, account) +} + +func TestDuplicateAccountDoesNotCopyCodexFingerprintSeed(t *testing.T) { + ctx := context.Background() + repo := &codexSeedDuplicateRepo{upstreamBillingProbeAccountRepo: &upstreamBillingProbeAccountRepo{accounts: make(map[int64]*Account)}} + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + source := &Account{ + Name: "source", + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: testCodexFingerprintSeed, + }, + } + require.NoError(t, repo.Create(ctx, source)) + + duplicate, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "") + + require.NoError(t, err) + require.NotEqual(t, source.ID, duplicate.ID) + require.NotContains(t, duplicate.Extra, codexFingerprintSeedExtraKey) + require.Equal(t, "session", duplicate.Extra[codexFingerprintModeExtraKey]) +} + +func TestDuplicateCreatePathMintsFreshSeedWhenEligible(t *testing.T) { + extra, err := duplicateAccountExtra(map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: testCodexFingerprintSeed, + }) + require.NoError(t, err) + + account, err := buildAccountForCreate(&CreateAccountInput{ + Name: "eligible-copy", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Extra: extra, + }, extra) + + require.NoError(t, err) + require.NotEqual(t, testCodexFingerprintSeed, requireValidCodexFingerprintSeed(t, account.Extra)) + require.Equal(t, "session", account.Extra[codexFingerprintModeExtraKey]) +} + +func TestAccountServiceCreateAndUpdateCodexSeedLifecycle(t *testing.T) { + ctx := context.Background() + repo := &upstreamBillingProbeAccountRepo{accounts: make(map[int64]*Account)} + svc := NewAccountService(repo, nil) + + created, err := svc.Create(ctx, CreateAccountRequest{ + Name: "legacy-create", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: userSuppliedCodexFingerprintSeed, + }, + }) + require.NoError(t, err) + createdSeed := requireValidCodexFingerprintSeed(t, created.Extra) + require.NotEqual(t, userSuppliedCodexFingerprintSeed, createdSeed) + + updateSeed := userSuppliedCodexFingerprintSeed + updated, err := svc.Update(ctx, created.ID, UpdateAccountRequest{ + Extra: &map[string]any{ + codexFingerprintModeExtraKey: "full", + codexFingerprintSeedExtraKey: updateSeed, + }, + }) + require.NoError(t, err) + require.Equal(t, createdSeed, requireValidCodexFingerprintSeed(t, updated.Extra)) + require.Equal(t, "full", updated.Extra[codexFingerprintModeExtraKey]) +} diff --git a/backend/internal/service/account_service.go b/backend/internal/service/account_service.go index be199cb256..96689e5000 100644 --- a/backend/internal/service/account_service.go +++ b/backend/internal/service/account_service.go @@ -166,6 +166,9 @@ type AccountBulkUpdate struct { Credentials map[string]any Extra map[string]any ProbeEnabled *bool + // EnsureCodexFingerprintSeed asks the repository to atomically preserve an + // existing valid Codex fingerprint seed or create one for eligible rows. + EnsureCodexFingerprintSeed bool } // CreateAccountRequest 创建账号请求 @@ -233,7 +236,7 @@ func (s *AccountService) Create(ctx context.Context, req CreateAccountRequest) ( Platform: req.Platform, Type: req.Type, Credentials: SanitizeStoredCredentials(req.Platform, req.Credentials), - Extra: req.Extra, + Extra: prepareCodexFingerprintExtraForCreate(req.Platform, req.Type, req.Extra), ProxyID: req.ProxyID, Concurrency: req.Concurrency, Priority: req.Priority, @@ -336,7 +339,9 @@ func (s *AccountService) Update(ctx context.Context, id int64, req UpdateAccount delete(extra, OllamaCloudUsageSessionExtraKey) delete(extra, OllamaCloudUsageAutoRefreshExtraKey) delete(extra, OllamaCloudUsageSnapshotExtraKey) - account.Extra = extra + account.Extra = prepareCodexFingerprintExtraForUpdate(account, extra) + } else { + account.Extra = prepareCodexFingerprintExtraForUpdate(account, account.Extra) } if req.ProxyID != nil { diff --git a/backend/internal/service/account_test_service_openai_compact_test.go b/backend/internal/service/account_test_service_openai_compact_test.go index 5b47725084..aa2b5f70bd 100644 --- a/backend/internal/service/account_test_service_openai_compact_test.go +++ b/backend/internal/service/account_test_service_openai_compact_test.go @@ -284,7 +284,10 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactProbeIdentityMatc "chatgpt_account_id": "chatgpt-acc", }, // 收敛是显式 opt-in（#5610），这里显式开启以验证探测身份与真实流量同构。 - Extra: map[string]any{"codex_fingerprint_mode": "session"}, + Extra: map[string]any{ + "codex_fingerprint_mode": "session", + codexFingerprintSeedExtraKey: testCodexFingerprintSeed, + }, } repo := &snapshotUpdateAccountRepo{ stubOpenAIAccountRepo: stubOpenAIAccountRepo{accounts: []Account{account}}, @@ -304,10 +307,12 @@ func TestAccountTestService_TestAccountConnection_OpenAICompactProbeIdentityMatc require.NoError(t, svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact)) // 显式 session 收敛模式：出站身份 = 账号级收敛值 - converged := resolveConvergedSessionID(&account) + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + converged := resolveConvergedSessionID(seed) require.Equal(t, converged, upstream.lastReq.Header.Get("session-id")) require.Equal(t, converged, upstream.lastReq.Header.Get("session_id")) - require.Equal(t, resolveConvergedInstallationID(&account), upstream.lastReq.Header.Get("x-codex-installation-id"), + require.Equal(t, resolveConvergedInstallationID(&account, seed), upstream.lastReq.Header.Get("x-codex-installation-id"), "真实 Codex 每个请求必带 installation-id，探测不得缺失") require.NotContains(t, upstream.lastReq.Header.Get("session-id"), "probe_compact", "探测标识不得是可被上游一眼识别的字面量") diff --git a/backend/internal/service/admin_account.go b/backend/internal/service/admin_account.go index 9958846658..ebfe861f17 100644 --- a/backend/internal/service/admin_account.go +++ b/backend/internal/service/admin_account.go @@ -128,22 +128,24 @@ var duplicateAccountDiscardedExtraKeys = map[string]struct{}{ "drive_storage_limit": {}, "drive_storage_usage": {}, "drive_tier_updated_at": {}, - "codex_primary_used_percent": {}, - "codex_primary_reset_after_seconds": {}, - "codex_primary_window_minutes": {}, - "codex_secondary_used_percent": {}, - "codex_secondary_reset_after_seconds": {}, - "codex_secondary_window_minutes": {}, - "codex_primary_over_secondary_percent": {}, - "codex_usage_updated_at": {}, - "codex_5h_used_percent": {}, - "codex_5h_reset_after_seconds": {}, - "codex_5h_window_minutes": {}, - "codex_5h_reset_at": {}, - "codex_7d_used_percent": {}, - "codex_7d_reset_after_seconds": {}, - "codex_7d_window_minutes": {}, - "codex_7d_reset_at": {}, + // Codex fingerprint convergence uses a per-account random seed, never copied from another account. + codexFingerprintSeedExtraKey: {}, + "codex_primary_used_percent": {}, + "codex_primary_reset_after_seconds": {}, + "codex_primary_window_minutes": {}, + "codex_secondary_used_percent": {}, + "codex_secondary_reset_after_seconds": {}, + "codex_secondary_window_minutes": {}, + "codex_primary_over_secondary_percent": {}, + "codex_usage_updated_at": {}, + "codex_5h_used_percent": {}, + "codex_5h_reset_after_seconds": {}, + "codex_5h_window_minutes": {}, + "codex_5h_reset_at": {}, + "codex_7d_used_percent": {}, + "codex_7d_reset_after_seconds": {}, + "codex_7d_window_minutes": {}, + "codex_7d_reset_at": {}, } func duplicateAccountExtra(value map[string]any) (map[string]any, error) { @@ -404,6 +406,7 @@ func buildAccountForCreate(input *CreateAccountInput, accountExtra map[string]an delete(accountExtra, OllamaCloudUsageSessionExtraKey) delete(accountExtra, OllamaCloudUsageAutoRefreshExtraKey) delete(accountExtra, OllamaCloudUsageSnapshotExtraKey) + accountExtra = prepareCodexFingerprintExtraForCreate(input.Platform, input.Type, accountExtra) account := &Account{ Name: input.Name, Notes: normalizeAccountNotes(input.Notes), @@ -651,6 +654,7 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U normalizedExtra[key] = v } } + normalizedExtra = prepareCodexFingerprintExtraForUpdate(account, normalizedExtra) account.Extra = normalizedExtra if account.Platform == PlatformAntigravity && wasOveragesEnabled && !account.IsOveragesEnabled() { delete(account.Extra, "antigravity_credits_overages") // 清理旧版 overages 运行态 @@ -670,6 +674,9 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U ComputeQuotaResetAt(account.Extra) NormalizeFixedQuotaWindows(account.Extra) } + if input.Extra == nil { + account.Extra = prepareCodexFingerprintExtraForUpdate(account, account.Extra) + } if requestedRateSyncEnabledUpdate != nil && *requestedRateSyncEnabledUpdate { if requestedProbeEnabledUpdate != nil && !*requestedProbeEnabledUpdate { return nil, infraerrors.BadRequest( @@ -852,6 +859,7 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U // UpdateAccountExtra 仅对 Extra JSONB 做 key 级合并，避免覆盖其它运行态键 // （如 model_rate_limits / passive_usage_* 等）。 func (s *adminServiceImpl) UpdateAccountExtra(ctx context.Context, id int64, updates map[string]any) error { + updates = sanitizedCodexFingerprintExtraUpdates(updates) delete(updates, UpstreamBillingProbeEnabledExtraKey) delete(updates, UpstreamBillingRateSyncEnabledExtraKey) delete(updates, UpstreamBillingProbeExtraKey) @@ -877,6 +885,7 @@ func (s *adminServiceImpl) UpdateAccountExtra(ctx context.Context, id int64, upd // It merges credentials/extra keys instead of overwriting the whole object. func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUpdateAccountsInput) (*BulkUpdateAccountsResult, error) { // Managed probe/session state may only enter through dedicated typed endpoints. + input.Extra = sanitizedCodexFingerprintExtraUpdates(input.Extra) delete(input.Extra, UpstreamBillingProbeEnabledExtraKey) delete(input.Extra, UpstreamBillingRateSyncEnabledExtraKey) delete(input.Extra, UpstreamBillingProbeExtraKey) @@ -1027,9 +1036,10 @@ func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUp // Prepare bulk updates for columns and JSONB fields. repoUpdates := AccountBulkUpdate{ - Credentials: input.Credentials, - Extra: input.Extra, - ProbeEnabled: input.ProbeEnabled, + Credentials: input.Credentials, + Extra: input.Extra, + ProbeEnabled: input.ProbeEnabled, + EnsureCodexFingerprintSeed: ShouldEnsureCodexFingerprintSeedForExtraUpdates(input.Extra), } if input.ProbeEnabled != nil { if repoUpdates.Extra == nil { diff --git a/backend/internal/service/openai_codex_fingerprint.go b/backend/internal/service/openai_codex_fingerprint.go index c74669c00c..0f841fb2fe 100644 --- a/backend/internal/service/openai_codex_fingerprint.go +++ b/backend/internal/service/openai_codex_fingerprint.go @@ -5,6 +5,7 @@ import ( "encoding/binary" "encoding/json" "fmt" + "maps" "net/http" "strings" "time" @@ -30,20 +31,30 @@ func stageCodexFingerprintIDs(c *gin.Context, ids *codexFingerprintIDs) { } } -// applyStagedCodexFingerprintHeaders 读取 context 暂存的收敛 ID 并改写出站头。 -// 非透传与透传两个请求构造器共用本函数，防止应用语义漂移。仅 OAuth 账号 -// 生效（stale 键在账号类型混合 failover 下由该门挡住）。 -func applyStagedCodexFingerprintHeaders(c *gin.Context, account *Account, h http.Header) { +func stagedCodexFingerprintIDs(c *gin.Context, account *Account) *codexFingerprintIDs { if c == nil || account == nil || account.Type != AccountTypeOAuth { - return + return nil } value, ok := c.Get(codexFingerprintIDsContextKey) if !ok { - return + return nil } - if ids, ok := value.(*codexFingerprintIDs); ok { - applyCodexFingerprintHeaders(h, ids) + ids, ok := value.(*codexFingerprintIDs) + if !ok || ids == nil || ids.accountID != account.ID { + return nil } + return ids +} + +// applyStagedCodexFingerprintHeaders 读取 context 暂存的收敛 ID 并改写出站头。 +// 非透传与透传两个请求构造器共用本函数，防止应用语义漂移。仅解析该 +// snapshot 的 OAuth 账号可读取，避免 stale context 跨账号 failover 泄漏。 +func applyStagedCodexFingerprintHeaders(c *gin.Context, account *Account, h http.Header) { + applyCodexFingerprintHeaders(h, stagedCodexFingerprintIDs(c, account)) +} + +func applyStagedCodexFingerprintClientMetadata(c *gin.Context, account *Account, reqBody map[string]any) bool { + return applyCodexFingerprintClientMetadata(reqBody, stagedCodexFingerprintIDs(c, account)) } // codexFingerprintMode 控制 OAuth 账号出站请求的设备指纹收敛强度。 @@ -68,7 +79,117 @@ const ( codexFingerprintFull codexFingerprintMode = "full" ) -const codexFingerprintModeExtraKey = "codex_fingerprint_mode" +const ( + codexFingerprintModeExtraKey = "codex_fingerprint_mode" + codexFingerprintSeedExtraKey = "codex_fingerprint_seed" +) + +func canonicalCodexFingerprintSeed(value any) (string, bool) { + raw, ok := value.(string) + if !ok { + return "", false + } + trimmed := strings.TrimSpace(raw) + parsed, err := uuid.Parse(trimmed) + if err != nil || parsed == uuid.Nil || trimmed != parsed.String() { + return "", false + } + return trimmed, true +} + +func newCodexFingerprintSeed() string { + return uuid.NewString() +} + +func stripCodexFingerprintSeed(extra map[string]any) map[string]any { + if extra == nil { + return nil + } + stripped := maps.Clone(extra) + delete(stripped, codexFingerprintSeedExtraKey) + return stripped +} + +func codexFingerprintModeFromExtra(extra map[string]any) codexFingerprintMode { + if extra == nil { + return codexFingerprintOff + } + raw, _ := extra[codexFingerprintModeExtraKey].(string) + switch codexFingerprintMode(strings.TrimSpace(raw)) { + case codexFingerprintOff, codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull: + return codexFingerprintMode(strings.TrimSpace(raw)) + default: + return codexFingerprintOff + } +} + +func codexFingerprintModeRequiresSeed(mode codexFingerprintMode) bool { + switch mode { + case codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull: + return true + default: + return false + } +} + +func codexFingerprintSeed(extra map[string]any) (string, bool) { + if extra == nil { + return "", false + } + return canonicalCodexFingerprintSeed(extra[codexFingerprintSeedExtraKey]) +} + +func prepareCodexFingerprintExtraForCreate(platform, accountType string, extra map[string]any) map[string]any { + prepared := stripCodexFingerprintSeed(extra) + if platform != PlatformOpenAI || accountType != AccountTypeOAuth || !codexFingerprintModeRequiresSeed(codexFingerprintModeFromExtra(prepared)) { + return prepared + } + if prepared == nil { + prepared = make(map[string]any, 1) + } + prepared[codexFingerprintSeedExtraKey] = newCodexFingerprintSeed() + return prepared +} + +func prepareCodexFingerprintExtraForUpdate(account *Account, extra map[string]any) map[string]any { + prepared := stripCodexFingerprintSeed(extra) + if account == nil || account.Platform != PlatformOpenAI || account.Type != AccountTypeOAuth { + return prepared + } + if seed, ok := codexFingerprintSeed(account.Extra); ok { + if prepared == nil { + prepared = make(map[string]any, 1) + } + prepared[codexFingerprintSeedExtraKey] = seed + return prepared + } + if codexFingerprintModeRequiresSeed(codexFingerprintModeFromExtra(prepared)) { + if prepared == nil { + prepared = make(map[string]any, 1) + } + prepared[codexFingerprintSeedExtraKey] = newCodexFingerprintSeed() + } + return prepared +} + +func sanitizedCodexFingerprintExtraUpdates(updates map[string]any) map[string]any { + if updates == nil { + return nil + } + sanitized := maps.Clone(updates) + delete(sanitized, codexFingerprintSeedExtraKey) + return sanitized +} + +// ShouldEnsureCodexFingerprintSeedForExtraUpdates reports whether a JSONB key-level +// extra update is enabling Codex fingerprint convergence and therefore must atomically +// preserve or create the system-managed per-account seed in the repository update. +func ShouldEnsureCodexFingerprintSeedForExtraUpdates(updates map[string]any) bool { + if updates == nil { + return false + } + return codexFingerprintModeRequiresSeed(codexFingerprintModeFromExtra(updates)) +} // GetCodexFingerprintMode 从账号 extra JSON 读取指纹收敛模式。 // @@ -85,13 +206,7 @@ func (a *Account) GetCodexFingerprintMode() codexFingerprintMode { if a == nil || !a.IsOpenAIOAuth() { return codexFingerprintOff } - raw := strings.TrimSpace(a.GetExtraString(codexFingerprintModeExtraKey)) - switch codexFingerprintMode(raw) { - case codexFingerprintOff, codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull: - return codexFingerprintMode(raw) - default: - return codexFingerprintOff - } + return codexFingerprintModeFromExtra(a.Extra) } // deriveStableUUIDv4 从种子确定性派生一个 UUIDv4 格式的字符串。 @@ -110,45 +225,53 @@ func deriveStableUUIDv4(seed string) string { } // resolveConvergedInstallationID 返回账号级恒定的 installation_id。 -// 优先使用管理员配置的真实 device_id，无则从 accountID 确定性派生。 -func resolveConvergedInstallationID(account *Account) string { +// 优先使用管理员配置的真实 device_id，无则从系统管理的账号随机种子确定性派生。 +func resolveConvergedInstallationID(account *Account, seed string) string { if account == nil { return "" } if deviceID := account.GetOpenAIDeviceID(); deviceID != "" { return deviceID } - return deriveStableUUIDv4(fmt.Sprintf("sub2api:codex-install-id:v1:%d", account.ID)) + if seed == "" { + return "" + } + return deriveStableUUIDv4("sub2api:codex-install-id:v2:" + seed) } // resolveConvergedSessionID 返回账号级恒定的 session_id。 -func resolveConvergedSessionID(account *Account) string { - if account == nil { +func resolveConvergedSessionID(seed string) string { + if seed == "" { return "" } - return deriveStableUUIDv4(fmt.Sprintf("sub2api:codex-session-id:v1:%d", account.ID)) + return deriveStableUUIDv4("sub2api:codex-session-id:v2:" + seed) } // resolveConvergedThreadID 按客户端原始 session-id 确定性派生 thread_id。 // 每个真实 Codex 会话（不同客户端启动实例）获得一个独立线程， // 模拟正常用户 spawn 子代理或开多窗口的模式。 -func resolveConvergedThreadID(account *Account, clientSessionID string) string { - if account == nil || clientSessionID == "" { +func resolveConvergedThreadID(seed, clientSessionID string) string { + if seed == "" || clientSessionID == "" { return "" } - return deriveStableUUIDv4(fmt.Sprintf("sub2api:codex-thread-id:v1:%d:%s", account.ID, clientSessionID)) + return deriveStableUUIDv4("sub2api:codex-thread-id:v2:" + seed + ":" + clientSessionID) } // codexFingerprintIDs 收敛后的完整 ID 集合。 // 由 resolveCodexFingerprintIDs 一次性生成，同一个实例在头改写和体改写之间共享， -// 确保所有载体中的 turn_id 等随机字段一致。 +// 确保所有载体中的 turn_id 等随机字段一致。体改写时还会补记原始 +// client_metadata.session_id，用于识别 root prompt_cache_key 的默认值。 type codexFingerprintIDs struct { - mode codexFingerprintMode - installationID string - sessionID string - threadID string - turnID string - windowID string + accountID int64 + mode codexFingerprintMode + installationID string + sessionID string + threadID string + turnID string + windowID string + turnStartedAtUnixMs int64 + originalBodySessionID string + originalBodySessionIDCaptured bool } // resolveCodexFingerprintIDs 按收敛模式计算出站 ID 集合。 @@ -157,13 +280,21 @@ type codexFingerprintIDs struct { // 返回 nil 表示 off 模式，不需要改写。 // 注意：包含随机生成的 turn_id，调用方必须只调用一次并共享结果给头改写和体改写。 func resolveCodexFingerprintIDs(account *Account, clientSessionID string, mode codexFingerprintMode) *codexFingerprintIDs { - if mode == codexFingerprintOff { + if account == nil || mode == codexFingerprintOff { + return nil + } + seed, ok := codexFingerprintSeed(account.Extra) + if !ok { return nil } - ids := &codexFingerprintIDs{mode: mode} + ids := &codexFingerprintIDs{ + accountID: account.ID, + mode: mode, + turnStartedAtUnixMs: time.Now().UnixMilli(), + } - ids.installationID = resolveConvergedInstallationID(account) + ids.installationID = resolveConvergedInstallationID(account, seed) if ids.installationID == "" { return nil } @@ -173,8 +304,8 @@ func resolveCodexFingerprintIDs(account *Account, clientSessionID string, mode c return ids case codexFingerprintSession: - ids.sessionID = resolveConvergedSessionID(account) - ids.threadID = resolveConvergedThreadID(account, clientSessionID) + ids.sessionID = resolveConvergedSessionID(seed) + ids.threadID = resolveConvergedThreadID(seed, clientSessionID) if ids.threadID == "" { ids.threadID = ids.sessionID } @@ -183,7 +314,7 @@ func resolveCodexFingerprintIDs(account *Account, clientSessionID string, mode c return ids case codexFingerprintFull: - ids.sessionID = resolveConvergedSessionID(account) + ids.sessionID = resolveConvergedSessionID(seed) ids.threadID = ids.sessionID ids.turnID = uuid.Must(uuid.NewV7()).String() ids.windowID = ids.threadID + ":0" @@ -252,20 +383,21 @@ func applyCodexFingerprintHeaders(h http.Header, ids *codexFingerprintIDs) { "thread_id": ids.threadID, "turn_id": ids.turnID, "window_id": ids.windowID, - "turn_started_at_unix_ms": time.Now().UnixMilli(), + "turn_started_at_unix_ms": ids.turnStartedAtUnixMs, }) } // rewriteCodexTurnMetadataFields 解析 x-codex-turn-metadata 头中的 JSON， -// 替换指定字段后回写。保留未指定字段原样（如 sandbox、thread_source 等）。 +// 替换指定字段后回写。合法对象保留未指定字段（如 sandbox、thread_source）； +// 非法/非对象值重建为最小合法 metadata，避免 flat 与 embedded identity 分裂。 func rewriteCodexTurnMetadataFields(h http.Header, fields map[string]any) { raw := strings.TrimSpace(h.Get("x-codex-turn-metadata")) if raw == "" { return } var metadata map[string]any - if err := json.Unmarshal([]byte(raw), &metadata); err != nil { - return + if err := json.Unmarshal([]byte(raw), &metadata); err != nil || metadata == nil { + metadata = make(map[string]any, len(fields)) } for k, v := range fields { metadata[k] = v @@ -284,16 +416,21 @@ func applyCodexFingerprintClientMetadata(reqBody map[string]any, ids *codexFinge return false } + captureCodexFingerprintOriginalBodySessionID(ids, reqBody["client_metadata"]) existing, _ := reqBody["client_metadata"].(map[string]any) if existing == nil { existing = make(map[string]any) } - if !applyCodexFingerprintToClientMetadataMap(existing, ids) { - return false + modified := false + if applyCodexFingerprintToClientMetadataMap(existing, ids) { + reqBody["client_metadata"] = existing + modified = true } - reqBody["client_metadata"] = existing - return true + if applyCodexFingerprintPromptCacheKey(reqBody, ids) { + modified = true + } + return modified } // applyCodexFingerprintToClientMetadataMap 是 client_metadata 改写的共享核心， @@ -330,59 +467,130 @@ func applyCodexFingerprintToClientMetadataMap(existing map[string]any, ids *code "thread_id": ids.threadID, "turn_id": ids.turnID, "window_id": ids.windowID, - "turn_started_at_unix_ms": time.Now().UnixMilli(), + "turn_started_at_unix_ms": ids.turnStartedAtUnixMs, }) return true } +func captureCodexFingerprintOriginalBodySessionID(ids *codexFingerprintIDs, clientMetadata any) { + if ids == nil || ids.originalBodySessionIDCaptured { + return + } + ids.originalBodySessionIDCaptured = true + if clientMetadata == nil { + return + } + switch metadata := clientMetadata.(type) { + case map[string]any: + if sessionID, ok := metadata["session_id"].(string); ok { + ids.originalBodySessionID = strings.TrimSpace(sessionID) + } + case map[string]string: + ids.originalBodySessionID = strings.TrimSpace(metadata["session_id"]) + } +} + +func captureCodexFingerprintOriginalBodySessionIDRaw(ids *codexFingerprintIDs, value gjson.Result) { + if ids == nil || ids.originalBodySessionIDCaptured { + return + } + ids.originalBodySessionIDCaptured = true + if value.Exists() && value.Type == gjson.String { + ids.originalBodySessionID = strings.TrimSpace(value.String()) + } +} + +func shouldRewriteCodexFingerprintPromptCacheKey(ids *codexFingerprintIDs, promptCacheKey string) bool { + if ids == nil || !ids.originalBodySessionIDCaptured || ids.originalBodySessionID == "" || ids.sessionID == "" { + return false + } + if ids.mode != codexFingerprintSession && ids.mode != codexFingerprintFull { + return false + } + return promptCacheKey == ids.originalBodySessionID +} + +func applyCodexFingerprintPromptCacheKey(reqBody map[string]any, ids *codexFingerprintIDs) bool { + if reqBody == nil { + return false + } + promptCacheKey, ok := reqBody["prompt_cache_key"].(string) + if !ok || strings.TrimSpace(promptCacheKey) == "" || !shouldRewriteCodexFingerprintPromptCacheKey(ids, promptCacheKey) { + return false + } + if promptCacheKey == ids.sessionID { + return false + } + reqBody["prompt_cache_key"] = ids.sessionID + return true +} + // applyCodexFingerprintClientMetadataRaw 在原始 JSON 字节上改写 client_metadata， // 供透传路径使用——透传是热路径，禁止对可能高达数十 MB 的 body 做全量 // Unmarshal（见 forwardOpenAIPassthrough 的轻量提取注释）。实现为：gjson 提取 // client_metadata 小对象单独解码，经共享核心改写后 sjson 一次性拼回，body -// 其余字节原样保留。语义与 applyCodexFingerprintClientMetadata 逐点一致 -// （含"非对象值整体替换为收敛集合"的行为）。 +// 其余字节原样保留；root prompt_cache_key 仅在可证明是 body session 默认值时 +// 做标量改写。语义与 applyCodexFingerprintClientMetadata 逐点一致（含 +// "非对象值整体替换为收敛集合"的行为）。 func applyCodexFingerprintClientMetadataRaw(body []byte, ids *codexFingerprintIDs) ([]byte, bool, error) { if len(body) == 0 || ids == nil { return body, false, nil } // 非 JSON 对象的 body（数组/标量/畸形）没有 client_metadata 语义， // sjson 在这类根上写字段会改写整体结构，直接放行保持原样。 - if !gjson.ParseBytes(body).IsObject() { + root := gjson.ParseBytes(body) + if !root.IsObject() { + captureCodexFingerprintOriginalBodySessionIDRaw(ids, gjson.Result{}) return body, false, nil } existing := map[string]any{} if cm := gjson.GetBytes(body, "client_metadata"); cm.IsObject() { + captureCodexFingerprintOriginalBodySessionIDRaw(ids, gjson.GetBytes(body, "client_metadata.session_id")) if err := json.Unmarshal([]byte(cm.Raw), &existing); err != nil { return body, false, fmt.Errorf("decode client_metadata for fingerprint: %w", err) } + } else { + captureCodexFingerprintOriginalBodySessionIDRaw(ids, gjson.Result{}) } - if !applyCodexFingerprintToClientMetadataMap(existing, ids) { - return body, false, nil + next := body + modified := false + if applyCodexFingerprintToClientMetadataMap(existing, ids) { + raw, err := json.Marshal(existing) + if err != nil { + return body, false, fmt.Errorf("encode converged client_metadata: %w", err) + } + var setErr error + next, setErr = sjson.SetRawBytes(body, "client_metadata", raw) + if setErr != nil { + return body, false, fmt.Errorf("splice converged client_metadata: %w", setErr) + } + modified = true } - - raw, err := json.Marshal(existing) - if err != nil { - return body, false, fmt.Errorf("encode converged client_metadata: %w", err) + promptCacheKey := gjson.GetBytes(body, "prompt_cache_key") + if promptCacheKey.Exists() && promptCacheKey.Type == gjson.String && strings.TrimSpace(promptCacheKey.String()) != "" && shouldRewriteCodexFingerprintPromptCacheKey(ids, promptCacheKey.String()) { + rewritten, err := sjson.SetBytes(next, "prompt_cache_key", ids.sessionID) + if err != nil { + return body, false, fmt.Errorf("splice converged prompt_cache_key: %w", err) + } + next = rewritten + modified = true } - next, err := sjson.SetRawBytes(body, "client_metadata", raw) - if err != nil { - return body, false, fmt.Errorf("splice converged client_metadata: %w", err) - } - return next, true, nil + return next, modified, nil } // rewriteClientMetadataEmbeddedTurnMetadata 改写 client_metadata 中内嵌的 -// x-codex-turn-metadata JSON 字符串里的指定字段。 +// x-codex-turn-metadata JSON 字符串里的指定字段。非法/非对象值会重建， +// 避免 flat client_metadata 与 embedded metadata 暴露两套身份。 func rewriteClientMetadataEmbeddedTurnMetadata(clientMetadata map[string]any, fields map[string]any) { raw, ok := clientMetadata["x-codex-turn-metadata"].(string) if !ok || raw == "" { return } var metadata map[string]any - if err := json.Unmarshal([]byte(raw), &metadata); err != nil { - return + if err := json.Unmarshal([]byte(raw), &metadata); err != nil || metadata == nil { + metadata = make(map[string]any, len(fields)) } for k, v := range fields { metadata[k] = v diff --git a/backend/internal/service/openai_codex_fingerprint_test.go b/backend/internal/service/openai_codex_fingerprint_test.go index 8e74bfa8bf..841ee3437d 100644 --- a/backend/internal/service/openai_codex_fingerprint_test.go +++ b/backend/internal/service/openai_codex_fingerprint_test.go @@ -13,7 +13,17 @@ import ( "github.com/stretchr/testify/require" ) +const testCodexFingerprintSeed = "11111111-1111-4111-8111-111111111111" + func newTestOAuthAccount(id int64, extra map[string]any) *Account { + if codexFingerprintModeRequiresSeed(codexFingerprintModeFromExtra(extra)) { + if extra == nil { + extra = make(map[string]any) + } + if _, exists := extra[codexFingerprintSeedExtraKey]; !exists { + extra[codexFingerprintSeedExtraKey] = testCodexFingerprintSeed + } + } return &Account{ ID: id, Platform: PlatformOpenAI, @@ -75,42 +85,40 @@ func TestGetCodexFingerprintMode(t *testing.T) { func TestResolveConvergedInstallationID_UsesDeviceID(t *testing.T) { account := newTestOAuthAccount(1, map[string]any{"openai_device_id": "real-device-id"}) - assert.Equal(t, "real-device-id", resolveConvergedInstallationID(account)) + assert.Equal(t, "real-device-id", resolveConvergedInstallationID(account, testCodexFingerprintSeed)) } -func TestResolveConvergedInstallationID_DerivesFromAccountID(t *testing.T) { +func TestResolveConvergedInstallationID_DerivesFromSeed(t *testing.T) { account := newTestOAuthAccount(42, nil) - result := resolveConvergedInstallationID(account) + result := resolveConvergedInstallationID(account, testCodexFingerprintSeed) _, err := uuid.Parse(result) require.NoError(t, err, "派生值应为合法 UUID") - assert.Equal(t, result, resolveConvergedInstallationID(account), "确定性") + assert.Equal(t, result, resolveConvergedInstallationID(account, testCodexFingerprintSeed), "确定性") } -func TestResolveConvergedInstallationID_DifferentAccounts(t *testing.T) { - a := resolveConvergedInstallationID(newTestOAuthAccount(1, nil)) - b := resolveConvergedInstallationID(newTestOAuthAccount(2, nil)) +func TestResolveConvergedInstallationID_DifferentSeeds(t *testing.T) { + account := newTestOAuthAccount(1, nil) + a := resolveConvergedInstallationID(account, testCodexFingerprintSeed) + b := resolveConvergedInstallationID(account, "22222222-2222-4222-8222-222222222222") assert.NotEqual(t, a, b) } // --- resolveConvergedThreadID --- func TestResolveConvergedThreadID_PerClientSession(t *testing.T) { - account := newTestOAuthAccount(1, nil) - a := resolveConvergedThreadID(account, "session-aaa") - b := resolveConvergedThreadID(account, "session-bbb") + a := resolveConvergedThreadID(testCodexFingerprintSeed, "session-aaa") + b := resolveConvergedThreadID(testCodexFingerprintSeed, "session-bbb") assert.NotEqual(t, a, b, "不同客户端 session 应得到不同 thread_id") } func TestResolveConvergedThreadID_Deterministic(t *testing.T) { - account := newTestOAuthAccount(1, nil) - a := resolveConvergedThreadID(account, "session-aaa") - b := resolveConvergedThreadID(account, "session-aaa") + a := resolveConvergedThreadID(testCodexFingerprintSeed, "session-aaa") + b := resolveConvergedThreadID(testCodexFingerprintSeed, "session-aaa") assert.Equal(t, a, b, "同一客户端 session 应得到相同 thread_id") } func TestResolveConvergedThreadID_EmptySession(t *testing.T) { - account := newTestOAuthAccount(1, nil) - assert.Equal(t, "", resolveConvergedThreadID(account, "")) + assert.Equal(t, "", resolveConvergedThreadID(testCodexFingerprintSeed, "")) } // --- off 模式：resolveCodexFingerprintIDsFromRequest 返回 nil --- @@ -141,6 +149,25 @@ func TestResolveCodexFingerprintIDsFromRequest_ExplicitOptInHonored(t *testing.T } } +func TestResolveCodexFingerprintIDsFromRequest_EnabledModesRequireValidSeed(t *testing.T) { + for _, tt := range []struct { + name string + extra map[string]any + }{ + {name: "missing", extra: map[string]any{codexFingerprintModeExtraKey: "device"}}, + {name: "missing with device override", extra: map[string]any{codexFingerprintModeExtraKey: "device", "openai_device_id": "real-device"}}, + {name: "blank", extra: map[string]any{codexFingerprintModeExtraKey: "session", codexFingerprintSeedExtraKey: ""}}, + {name: "uppercase", extra: map[string]any{codexFingerprintModeExtraKey: "full", codexFingerprintSeedExtraKey: "11111111-1111-4111-8111-AAAAAAAAAAAA"}}, + {name: "nil uuid", extra: map[string]any{codexFingerprintModeExtraKey: "device", codexFingerprintSeedExtraKey: "00000000-0000-0000-0000-000000000000"}}, + {name: "non string", extra: map[string]any{codexFingerprintModeExtraKey: "session", codexFingerprintSeedExtraKey: 123}}, + } { + t.Run(tt.name, func(t *testing.T) { + account := &Account{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: tt.extra} + require.Nil(t, resolveCodexFingerprintIDsFromRequest(account, nil)) + }) + } +} + // --- applyCodexFingerprintHeaders: off 模式 --- func TestApplyCodexFingerprintHeaders_OffMode(t *testing.T) { @@ -199,9 +226,11 @@ func TestApplyCodexFingerprintHeaders_SessionMode(t *testing.T) { ids := resolveCodexFingerprintIDsFromRequest(account, clientHeaders) applyCodexFingerprintHeaders(h, ids) - convergedInstall := resolveConvergedInstallationID(account) - convergedSession := resolveConvergedSessionID(account) - convergedThread := resolveConvergedThreadID(account, "client-session-aaa") + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + convergedInstall := resolveConvergedInstallationID(account, seed) + convergedSession := resolveConvergedSessionID(seed) + convergedThread := resolveConvergedThreadID(seed, "client-session-aaa") assert.Equal(t, convergedInstall, h.Get("x-codex-installation-id")) assert.Equal(t, convergedSession, h.Get("session-id")) @@ -257,7 +286,9 @@ func TestApplyCodexFingerprintHeaders_FullMode(t *testing.T) { account := newTestOAuthAccount(1, map[string]any{ codexFingerprintModeExtraKey: "full", }) - convergedSession := resolveConvergedSessionID(account) + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + convergedSession := resolveConvergedSessionID(seed) clientA := http.Header{} clientA.Set("session-id", "client-A") @@ -329,6 +360,41 @@ func TestFingerprintIDs_HeaderAndBody_TurnID_Consistent(t *testing.T) { assert.Equal(t, headerTurnID, bodyTurnID, "头和体的 turn_id 必须一致") assert.Equal(t, headerTurnID, bodyEmbeddedTurnID, "头和体内嵌 turn-metadata 的 turn_id 必须一致") assert.Equal(t, ids.turnID, headerTurnID, "所有 turn_id 都应来自同一份 ids") + assert.Equal(t, headerMeta["turn_started_at_unix_ms"], bodyMeta["turn_started_at_unix_ms"], "头和体的 timestamp 必须一致") + assert.Equal(t, float64(ids.turnStartedAtUnixMs), headerMeta["turn_started_at_unix_ms"]) +} + +func TestFingerprintIDs_MalformedEmbeddedMetadataRebuiltConsistently(t *testing.T) { + account := newTestOAuthAccount(2, map[string]any{codexFingerprintModeExtraKey: "session"}) + clientHeaders := make(http.Header) + clientHeaders.Set("session-id", "client-session-malformed") + ids := resolveCodexFingerprintIDsFromRequest(account, clientHeaders) + require.NotNil(t, ids) + + h := make(http.Header) + h.Set("x-codex-turn-metadata", "{malformed") + applyCodexFingerprintHeaders(h, ids) + + reqBody := map[string]any{ + "client_metadata": map[string]any{ + "session_id": "client-session-malformed", + "x-codex-turn-metadata": "[malformed", + }, + } + require.True(t, applyCodexFingerprintClientMetadata(reqBody, ids)) + + var headerMeta map[string]any + require.NoError(t, json.Unmarshal([]byte(h.Get("x-codex-turn-metadata")), &headerMeta)) + clientMetadata, ok := reqBody["client_metadata"].(map[string]any) + require.True(t, ok) + bodyRaw, ok := clientMetadata["x-codex-turn-metadata"].(string) + require.True(t, ok) + var bodyMeta map[string]any + require.NoError(t, json.Unmarshal([]byte(bodyRaw), &bodyMeta)) + + for _, key := range []string{"installation_id", "session_id", "thread_id", "turn_id", "window_id", "turn_started_at_unix_ms"} { + assert.Equal(t, headerMeta[key], bodyMeta[key], "rebuilt metadata field %s must match", key) + } } // --- applyCodexFingerprintClientMetadata --- @@ -400,9 +466,11 @@ func TestApplyCodexFingerprintClientMetadata_SessionMode(t *testing.T) { cm, ok := reqBody["client_metadata"].(map[string]any) require.True(t, ok) - convergedInstall := resolveConvergedInstallationID(account) - convergedSession := resolveConvergedSessionID(account) - convergedThread := resolveConvergedThreadID(account, "client-session-aaa") + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + convergedInstall := resolveConvergedInstallationID(account, seed) + convergedSession := resolveConvergedSessionID(seed) + convergedThread := resolveConvergedThreadID(seed, "client-session-aaa") assert.Equal(t, convergedInstall, cm["x-codex-installation-id"]) assert.Equal(t, convergedSession, cm["session_id"]) @@ -441,7 +509,9 @@ func TestApplyCodexFingerprintClientMetadata_FullMode(t *testing.T) { cm, ok := reqBody["client_metadata"].(map[string]any) require.True(t, ok) - convergedSession := resolveConvergedSessionID(account) + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + convergedSession := resolveConvergedSessionID(seed) assert.Equal(t, convergedSession, cm["session_id"]) assert.Equal(t, convergedSession, cm["thread_id"], "full 模式 thread_id 应等于 session_id") @@ -496,6 +566,182 @@ func rawVsMapClientMetadata(t *testing.T, body []byte, ids *codexFingerprintIDs) return mapCM, rawCM } +func cloneCodexFingerprintIDsForTest(ids *codexFingerprintIDs) *codexFingerprintIDs { + if ids == nil { + return nil + } + cloned := *ids + cloned.originalBodySessionID = "" + cloned.originalBodySessionIDCaptured = false + return &cloned +} + +func applyMapAndRawFingerprintBodiesForTest(t *testing.T, body []byte, ids *codexFingerprintIDs) (map[string]any, map[string]any) { + t.Helper() + + mapIDs := cloneCodexFingerprintIDsForTest(ids) + rawIDs := cloneCodexFingerprintIDsForTest(ids) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(body, &decoded)) + applyCodexFingerprintClientMetadata(decoded, mapIDs) + + rawBody, _, err := applyCodexFingerprintClientMetadataRaw(body, rawIDs) + require.NoError(t, err) + var rawDecoded map[string]any + require.NoError(t, json.Unmarshal(rawBody, &rawDecoded)) + return decoded, rawDecoded +} + +func TestApplyCodexFingerprintPromptCacheKey_MapRawEquivalence(t *testing.T) { + for _, mode := range []codexFingerprintMode{codexFingerprintSession, codexFingerprintFull} { + t.Run(string(mode)+"/default", func(t *testing.T) { + account := newTestOAuthAccount(4300, map[string]any{codexFingerprintModeExtraKey: string(mode)}) + ids := resolveCodexFingerprintIDs(account, "header-session", mode) + require.NotNil(t, ids) + + body := []byte(`{"model":"gpt-5.6-sol","prompt_cache_key":"body-session","client_metadata":{"session_id":" body-session ","trace":"keep"},"input":[]}`) + mapBody, rawBody := applyMapAndRawFingerprintBodiesForTest(t, body, ids) + + require.Equal(t, mapBody["prompt_cache_key"], rawBody["prompt_cache_key"]) + require.Equal(t, ids.sessionID, mapBody["prompt_cache_key"]) + mapCM, _ := mapBody["client_metadata"].(map[string]any) + rawCM, _ := rawBody["client_metadata"].(map[string]any) + require.Equal(t, ids.sessionID, mapCM["session_id"]) + require.Equal(t, mapCM["session_id"], rawCM["session_id"]) + require.Equal(t, "keep", rawCM["trace"]) + }) + } + + t.Run("explicit override", func(t *testing.T) { + account := newTestOAuthAccount(4301, map[string]any{codexFingerprintModeExtraKey: "session"}) + ids := resolveCodexFingerprintIDs(account, "header-session", codexFingerprintSession) + require.NotNil(t, ids) + + body := []byte(`{"model":"gpt-5.6-sol","prompt_cache_key":"explicit-cache","client_metadata":{"session_id":"body-session"},"input":[]}`) + mapBody, rawBody := applyMapAndRawFingerprintBodiesForTest(t, body, ids) + + require.Equal(t, "explicit-cache", mapBody["prompt_cache_key"]) + require.Equal(t, "explicit-cache", rawBody["prompt_cache_key"]) + mapCM, _ := mapBody["client_metadata"].(map[string]any) + rawCM, _ := rawBody["client_metadata"].(map[string]any) + require.Equal(t, ids.sessionID, mapCM["session_id"]) + require.Equal(t, ids.sessionID, rawCM["session_id"]) + }) +} + +func TestApplyCodexFingerprintPromptCacheKey_Negatives(t *testing.T) { + sessionAccount := newTestOAuthAccount(4310, map[string]any{codexFingerprintModeExtraKey: "session"}) + sessionIDs := resolveCodexFingerprintIDs(sessionAccount, "header-session", codexFingerprintSession) + require.NotNil(t, sessionIDs) + deviceAccount := newTestOAuthAccount(4311, map[string]any{codexFingerprintModeExtraKey: "device"}) + deviceIDs := resolveCodexFingerprintIDs(deviceAccount, "header-session", codexFingerprintDevice) + require.NotNil(t, deviceIDs) + + tests := []struct { + name string + body []byte + ids *codexFingerprintIDs + wantExists bool + wantCacheKey any + wantRawString string + }{ + { + name: "missing key is not injected", + body: []byte(`{"client_metadata":{"session_id":"body-session"}}`), + ids: sessionIDs, + wantExists: false, + }, + { + name: "empty key preserved", + body: []byte(`{"prompt_cache_key":"","client_metadata":{"session_id":"body-session"}}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: "", + }, + { + name: "whitespace-different key is an explicit override", + body: []byte(`{"prompt_cache_key":" body-session ","client_metadata":{"session_id":"body-session"}}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: " body-session ", + }, + { + name: "non-string key preserved", + body: []byte(`{"prompt_cache_key":123,"client_metadata":{"session_id":"body-session"}}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: float64(123), + }, + { + name: "missing source metadata preserves key", + body: []byte(`{"prompt_cache_key":"body-session"}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: "body-session", + }, + { + name: "non-string source session preserves key", + body: []byte(`{"prompt_cache_key":"123","client_metadata":{"session_id":123}}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: "123", + }, + { + name: "non-object source metadata preserves key", + body: []byte(`{"prompt_cache_key":"body-session","client_metadata":"bad"}`), + ids: sessionIDs, + wantExists: true, + wantCacheKey: "body-session", + }, + { + name: "device mode preserves key", + body: []byte(`{"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session"}}`), + ids: deviceIDs, + wantExists: true, + wantCacheKey: "body-session", + }, + { + name: "off mode preserves body", + body: []byte(`{"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session"}}`), + ids: nil, + wantExists: true, + wantCacheKey: "body-session", + wantRawString: `{"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session"}}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var mapBody map[string]any + require.NoError(t, json.Unmarshal(tt.body, &mapBody)) + changedMap := applyCodexFingerprintClientMetadata(mapBody, cloneCodexFingerprintIDsForTest(tt.ids)) + + rawBody, changedRaw, err := applyCodexFingerprintClientMetadataRaw(tt.body, cloneCodexFingerprintIDsForTest(tt.ids)) + require.NoError(t, err) + if tt.ids == nil { + require.False(t, changedMap) + require.False(t, changedRaw) + require.JSONEq(t, tt.wantRawString, string(rawBody)) + return + } + require.True(t, changedMap) + require.True(t, changedRaw) + + rawDecoded := map[string]any{} + require.NoError(t, json.Unmarshal(rawBody, &rawDecoded)) + _, mapExists := mapBody["prompt_cache_key"] + _, rawExists := rawDecoded["prompt_cache_key"] + require.Equal(t, tt.wantExists, mapExists) + require.Equal(t, tt.wantExists, rawExists) + if tt.wantExists { + require.Equal(t, tt.wantCacheKey, mapBody["prompt_cache_key"]) + require.Equal(t, tt.wantCacheKey, rawDecoded["prompt_cache_key"]) + } + }) + } +} + func TestApplyCodexFingerprintClientMetadataRaw_MatchesMapVariant(t *testing.T) { embedded := `{\"installation_id\":\"real-install\",\"session_id\":\"real-session\",\"sandbox\":\"seatbelt\"}` bodies := map[string]string{ @@ -504,7 +750,7 @@ func TestApplyCodexFingerprintClientMetadataRaw_MatchesMapVariant(t *testing.T) "non_object_value": `{"model":"gpt-5.6-sol","client_metadata":"bogus","stream":true}`, } for _, mode := range []codexFingerprintMode{codexFingerprintDevice, codexFingerprintSession, codexFingerprintFull} { - account := newTestOAuthAccount(4242, nil) + account := newTestOAuthAccount(4242, map[string]any{codexFingerprintModeExtraKey: string(mode)}) ids := resolveCodexFingerprintIDs(account, "client-sess-raw", mode) require.NotNil(t, ids) for name, body := range bodies { @@ -517,7 +763,7 @@ func TestApplyCodexFingerprintClientMetadataRaw_MatchesMapVariant(t *testing.T) } func TestApplyCodexFingerprintClientMetadataRaw_PreservesUnrelatedFields(t *testing.T) { - account := newTestOAuthAccount(4243, nil) + account := newTestOAuthAccount(4243, map[string]any{codexFingerprintModeExtraKey: "session"}) ids := resolveCodexFingerprintIDs(account, "client-sess-preserve", codexFingerprintSession) require.NotNil(t, ids) @@ -562,7 +808,7 @@ func newFingerprintStageTestContext(t *testing.T) *gin.Context { func TestStageCodexFingerprintIDs_NilOverwritesPreviousAccount(t *testing.T) { c := newFingerprintStageTestContext(t) - accountA := newTestOAuthAccount(1001, nil) + accountA := newTestOAuthAccount(1001, map[string]any{codexFingerprintModeExtraKey: "session"}) idsA := resolveCodexFingerprintIDs(accountA, "sess-x", codexFingerprintSession) require.NotNil(t, idsA) stageCodexFingerprintIDs(c, idsA) @@ -578,9 +824,30 @@ func TestStageCodexFingerprintIDs_NilOverwritesPreviousAccount(t *testing.T) { assert.Empty(t, h.Get("x-codex-installation-id")) } +func TestApplyStagedCodexFingerprintRejectsDifferentOAuthAccount(t *testing.T) { + c := newFingerprintStageTestContext(t) + accountA := newTestOAuthAccount(1003, map[string]any{codexFingerprintModeExtraKey: "session"}) + idsA := resolveCodexFingerprintIDs(accountA, "sess-a", codexFingerprintSession) + require.NotNil(t, idsA) + stageCodexFingerprintIDs(c, idsA) + + accountB := newTestOAuthAccount(1004, map[string]any{codexFingerprintModeExtraKey: "session"}) + h := make(http.Header) + h.Set("session-id", "account-b-session") + applyStagedCodexFingerprintHeaders(c, accountB, h) + assert.Equal(t, "account-b-session", h.Get("session-id")) + assert.Empty(t, h.Get("x-codex-installation-id")) + + body := map[string]any{"client_metadata": map[string]any{"session_id": "account-b-session"}} + assert.False(t, applyStagedCodexFingerprintClientMetadata(c, accountB, body)) + clientMetadata, ok := body["client_metadata"].(map[string]any) + require.True(t, ok) + assert.Equal(t, "account-b-session", clientMetadata["session_id"]) +} + func TestApplyStagedCodexFingerprintHeaders_SkipsNonOAuthAccount(t *testing.T) { c := newFingerprintStageTestContext(t) - oauthIDs := resolveCodexFingerprintIDs(newTestOAuthAccount(1003, nil), "sess-y", codexFingerprintSession) + oauthIDs := resolveCodexFingerprintIDs(newTestOAuthAccount(1003, map[string]any{codexFingerprintModeExtraKey: "session"}), "sess-y", codexFingerprintSession) require.NotNil(t, oauthIDs) stageCodexFingerprintIDs(c, oauthIDs) @@ -643,12 +910,12 @@ func TestBuildUpstreamRequestOpenAIPassthrough_OffModeKeepsIsolatedSession(t *te require.NoError(t, err) assert.NotEmpty(t, req.Header.Get("session_id")) - assert.NotEqual(t, resolveConvergedSessionID(account), req.Header.Get("session_id"), "off 模式不得收敛 session_id") + assert.NotEqual(t, resolveConvergedSessionID(testCodexFingerprintSeed), req.Header.Get("session_id"), "off 模式不得收敛 session_id") assert.Empty(t, req.Header.Get("x-codex-window-id")) } func TestApplyCodexFingerprintClientMetadataRaw_NonObjectBodyUntouched(t *testing.T) { - account := newTestOAuthAccount(4244, nil) + account := newTestOAuthAccount(4244, map[string]any{codexFingerprintModeExtraKey: "session"}) ids := resolveCodexFingerprintIDs(account, "client-sess-nonobj", codexFingerprintSession) require.NotNil(t, ids) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index a262fb60cc..6dab7fb702 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -421,6 +421,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if !isCompactRequest && applyCodexClientMetadata(decoded, account) { markDecodedModified() } + stageCodexFingerprintIDs(c, nil) // 指纹收敛：一次性解析收敛 ID，请求体和出站头共享同一份 IDs（保证 turn_id 等随机字段一致）。 // fingerprintIDs 在此处解析，后续 buildUpstreamRequest 中使用同一份。 if !isCompactRequest { @@ -442,7 +443,9 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if codexResult.NormalizedModel != "" { upstreamModel = codexResult.NormalizedModel } - if codexResult.PromptCacheKey != "" { + if currentPromptCacheKey, ok := decoded["prompt_cache_key"].(string); ok && currentPromptCacheKey != "" { + promptCacheKey = currentPromptCacheKey + } else if codexResult.PromptCacheKey != "" { promptCacheKey = codexResult.PromptCacheKey } } diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 291f8bcc9e..4bfd2fa26c 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -81,6 +81,7 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( } reqStream = gjson.GetBytes(body, "stream").Bool() + stageCodexFingerprintIDs(c, nil) // 指纹收敛：与非透传路径同门控（仅 OAuth、legacy compact 形态跳过）。 // 一次性解析收敛 ID：请求体 client_metadata 在此改写（raw 字节外科 // 手术，透传热路径禁全量 Unmarshal），出站头改写由请求构造器读取 diff --git a/backend/internal/service/openai_oauth_passthrough_test.go b/backend/internal/service/openai_oauth_passthrough_test.go index d2c030c5c5..c7dc77b3aa 100644 --- a/backend/internal/service/openai_oauth_passthrough_test.go +++ b/backend/internal/service/openai_oauth_passthrough_test.go @@ -1948,6 +1948,212 @@ func TestOpenAIGatewayService_OAuthPassthrough_CodexTuiIdentityUnified(t *testin require.Equal(t, codexCLIVersion, upstream.lastReq.Header.Get("version")) } +func TestOpenAIGatewayService_CodexFingerprintHTTPTransformedHeaderBodyParityAndDefaultCacheKey(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("session-id", "header-session") + c.Request.Header.Set("x-codex-turn-metadata", `{"installation_id":"header-install","session_id":"header-session","thread_id":"header-thread","turn_id":"header-turn","window_id":"header-window","sandbox":"seatbelt"}`) + + body := []byte(`{"model":"gpt-5.2","stream":false,"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session","x-codex-turn-metadata":"{\"installation_id\":\"body-install\",\"session_id\":\"body-session\",\"thread_id\":\"body-thread\",\"turn_id\":\"body-turn\",\"window_id\":\"body-window\",\"sandbox\":\"seatbelt\"}"},"input":[{"type":"message","role":"user","content":"hi"}]}`) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid"}}, + Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n")), + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{}, + httpUpstream: upstream, + toolCorrector: NewCodexToolCorrector(), + } + account := newTestOAuthAccount(4401, map[string]any{codexFingerprintModeExtraKey: "session"}) + account.Name = "oauth-transformed" + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 1 + account.Credentials = map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"} + + _, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, upstream.lastReq) + + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + wantInstall := resolveConvergedInstallationID(account, seed) + wantSession := resolveConvergedSessionID(seed) + wantThread := resolveConvergedThreadID(seed, "header-session") + + require.Equal(t, wantInstall, upstream.lastReq.Header.Get("x-codex-installation-id")) + require.Equal(t, wantSession, upstream.lastReq.Header.Get("session-id")) + require.Equal(t, wantSession, upstream.lastReq.Header.Get("session_id")) + require.Equal(t, wantThread, upstream.lastReq.Header.Get("thread-id")) + require.Equal(t, wantThread, upstream.lastReq.Header.Get("x-client-request-id")) + require.Equal(t, wantThread+":0", upstream.lastReq.Header.Get("x-codex-window-id")) + + require.Equal(t, wantSession, gjson.GetBytes(upstream.lastBody, "prompt_cache_key").String()) + require.Equal(t, wantInstall, gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-installation-id").String()) + require.Equal(t, wantSession, gjson.GetBytes(upstream.lastBody, "client_metadata.session_id").String()) + require.Equal(t, wantThread, gjson.GetBytes(upstream.lastBody, "client_metadata.thread_id").String()) + require.Equal(t, wantThread+":0", gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-window-id").String()) + + bodyTurnMetadata := gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-turn-metadata").String() + headerTurnMetadata := upstream.lastReq.Header.Get("x-codex-turn-metadata") + require.Equal(t, wantSession, gjson.Get(bodyTurnMetadata, "session_id").String()) + require.Equal(t, wantSession, gjson.Get(headerTurnMetadata, "session_id").String()) + require.Equal(t, gjson.Get(bodyTurnMetadata, "turn_id").String(), gjson.Get(headerTurnMetadata, "turn_id").String()) +} + +func TestOpenAIGatewayService_CodexFingerprintHTTPRawPassthroughHeaderBodyParityAndDefaultCacheKey(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("session-id", "header-session") + c.Request.Header.Set("x-codex-turn-metadata", `{"installation_id":"header-install","session_id":"header-session","thread_id":"header-thread","turn_id":"header-turn","window_id":"header-window","sandbox":"seatbelt"}`) + + body := []byte(`{"model":"gpt-5.6-sol","stream":false,"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session","x-codex-turn-metadata":"{\"installation_id\":\"body-install\",\"session_id\":\"body-session\",\"thread_id\":\"body-thread\",\"turn_id\":\"body-turn\",\"window_id\":\"body-window\",\"sandbox\":\"seatbelt\"}"},"input":[{"type":"message","role":"user","content":"hi"}]}`) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid"}}, + Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n")), + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{}, + httpUpstream: upstream, + } + account := newTestOAuthAccount(4402, map[string]any{ + codexFingerprintModeExtraKey: "session", + "openai_oauth_passthrough": true, + }) + account.Name = "oauth-raw" + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 1 + account.Credentials = map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"} + + _, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, upstream.lastReq) + + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + wantInstall := resolveConvergedInstallationID(account, seed) + wantSession := resolveConvergedSessionID(seed) + wantThread := resolveConvergedThreadID(seed, "header-session") + + require.Equal(t, wantInstall, upstream.lastReq.Header.Get("x-codex-installation-id")) + require.Equal(t, wantSession, upstream.lastReq.Header.Get("session-id")) + require.Equal(t, wantSession, upstream.lastReq.Header.Get("session_id")) + require.Equal(t, wantThread, upstream.lastReq.Header.Get("thread-id")) + require.Equal(t, wantThread, upstream.lastReq.Header.Get("x-client-request-id")) + require.Equal(t, wantThread+":0", upstream.lastReq.Header.Get("x-codex-window-id")) + + require.Equal(t, wantSession, gjson.GetBytes(upstream.lastBody, "prompt_cache_key").String()) + require.Equal(t, wantInstall, gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-installation-id").String()) + require.Equal(t, wantSession, gjson.GetBytes(upstream.lastBody, "client_metadata.session_id").String()) + require.Equal(t, wantThread, gjson.GetBytes(upstream.lastBody, "client_metadata.thread_id").String()) + require.Equal(t, wantThread+":0", gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-window-id").String()) + + bodyTurnMetadata := gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-turn-metadata").String() + headerTurnMetadata := upstream.lastReq.Header.Get("x-codex-turn-metadata") + require.Equal(t, wantSession, gjson.Get(bodyTurnMetadata, "session_id").String()) + require.Equal(t, wantSession, gjson.Get(headerTurnMetadata, "session_id").String()) + require.Equal(t, gjson.Get(bodyTurnMetadata, "turn_id").String(), gjson.Get(headerTurnMetadata, "turn_id").String()) +} + +func TestOpenAIGatewayService_CodexFingerprintCompactDoesNotRewriteBodyCacheKeyOrMetadata(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses/compact", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("session-id", "header-session") + + body := []byte(`{"model":"gpt-5.4","stream":false,"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session"},"input":[{"type":"message","role":"user","content":"compress"}]}`) + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid"}}, + Body: io.NopCloser(strings.NewReader(compactProbeSSESuccessBody)), + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{}, + httpUpstream: upstream, + toolCorrector: NewCodexToolCorrector(), + } + account := newTestOAuthAccount(4403, map[string]any{codexFingerprintModeExtraKey: "session"}) + account.Name = "oauth-compact" + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 1 + account.Credentials = map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"} + staleIDs := resolveCodexFingerprintIDs(account, "stale-session", codexFingerprintSession) + require.NotNil(t, staleIDs) + stageCodexFingerprintIDs(c, staleIDs) + + _, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, upstream.lastReq) + + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + require.NotEqual(t, resolveConvergedSessionID(seed), gjson.GetBytes(upstream.lastBody, "prompt_cache_key").String()) + require.Equal(t, "body-session", gjson.GetBytes(upstream.lastBody, "prompt_cache_key").String()) + require.Equal(t, "body-session", gjson.GetBytes(upstream.lastBody, "client_metadata.session_id").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "client_metadata.x-codex-installation-id").Exists()) + require.Empty(t, upstream.lastReq.Header.Get("x-codex-window-id")) +} + +func TestOpenAIGatewayService_CodexFingerprintMessagesBridgeDoesNotInjectBodyPromptCacheKey(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("session-id", "header-session") + + body := []byte(`{"model":"gpt-5.5","stream":true,"prompt_cache_key":"anthropic-metadata-session-1","client_metadata":{"session_id":"anthropic-metadata-session-1"},"input":[{"type":"message","role":"developer","content":[{"type":"input_text","text":"` + openAICompatClaudeCodeTodoGuardMarker + `"}]},{"type":"message","role":"user","content":"hello"}]}`) + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid"}}, + Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n")), + }} + svc := &OpenAIGatewayService{ + cfg: &config.Config{}, + httpUpstream: upstream, + toolCorrector: NewCodexToolCorrector(), + } + account := newTestOAuthAccount(4404, map[string]any{codexFingerprintModeExtraKey: "session"}) + account.Name = "oauth-messages-bridge" + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 1 + account.Credentials = map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"} + + _, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, upstream.lastReq) + + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + wantSession := resolveConvergedSessionID(seed) + require.False(t, gjson.GetBytes(upstream.lastBody, "prompt_cache_key").Exists()) + require.Equal(t, wantSession, gjson.GetBytes(upstream.lastBody, "client_metadata.session_id").String()) + require.Equal(t, wantSession, upstream.lastReq.Header.Get("session_id")) +} + func TestOpenAIGatewayService_CodexCLIOnly_RejectsNonCodexClient(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index d783c98104..797361a3a6 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -93,7 +93,13 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( headers.Add("x-codex-beta-features", value) } } - for _, name := range [...]string{"x-codex-window-id", "x-codex-installation-id"} { + for _, name := range [...]string{ + "x-codex-window-id", + "x-codex-installation-id", + "session-id", + "thread-id", + "x-client-request-id", + } { if value := c.Request.Header.Get(name); strings.TrimSpace(value) != "" { headers.Set(name, value) } @@ -128,6 +134,7 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( if metadata := strings.TrimSpace(turnMetadata); metadata != "" { headers.Set(openAIWSTurnMetadataHeader, metadata) } + applyStagedCodexFingerprintHeaders(c, account, headers) if account != nil && account.Type == AccountTypeOAuth { if err := resolveAndSetOpenAIChatGPTAccountHeaders(ctx, s.accountRepo, headers, account); err != nil { diff --git a/backend/internal/service/openai_ws_forwarder_success_test.go b/backend/internal/service/openai_ws_forwarder_success_test.go index d8270dfe40..2c69776382 100644 --- a/backend/internal/service/openai_ws_forwarder_success_test.go +++ b/backend/internal/service/openai_ws_forwarder_success_test.go @@ -400,6 +400,9 @@ func TestOpenAIGatewayService_BuildOpenAIWSHeadersPreservesCodexIdentity(t *test c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") c.Request.Header.Set("X-Codex-Window-ID", "window-ws") c.Request.Header.Set("X-Codex-Installation-ID", "installation-ws") + c.Request.Header.Set("session-id", "session-ws") + c.Request.Header.Set("thread-id", "thread-ws") + c.Request.Header.Set("x-client-request-id", "client-request-ws") c.Request.Header.Set("X-Test", "blocked") svc := &OpenAIGatewayService{} @@ -421,9 +424,53 @@ func TestOpenAIGatewayService_BuildOpenAIWSHeadersPreservesCodexIdentity(t *test require.NoError(t, err) require.Equal(t, "window-ws", headers.Get("X-Codex-Window-ID")) require.Equal(t, "installation-ws", headers.Get("X-Codex-Installation-ID")) + require.Equal(t, "session-ws", headers.Get("session-id")) + require.Equal(t, "thread-ws", headers.Get("thread-id")) + require.Equal(t, "client-request-ws", headers.Get("x-client-request-id")) require.Empty(t, headers.Get("X-Test")) } +func TestOpenAIGatewayService_BuildOpenAIWSHeadersDeviceModePreservesClientSessionIdentity(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + 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") + + account := newTestOAuthAccount(1300, map[string]any{codexFingerprintModeExtraKey: "device"}) + ids := resolveCodexFingerprintIDsFromRequest(account, c.Request.Header) + require.NotNil(t, ids) + stageCodexFingerprintIDs(c, ids) + + svc := &OpenAIGatewayService{} + headers, _, err := svc.buildOpenAIWSHeaders( + context.Background(), + c, + account, + "token", + OpenAIWSProtocolDecision{Transport: OpenAIUpstreamTransportResponsesWebsocketV2}, + true, + "", + "", + "", + "", + "", + ) + + 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")) +} + func TestLogOpenAIWSBindResponseAccountWarn(t *testing.T) { require.NotPanics(t, func() { logOpenAIWSBindResponseAccountWarn(1, 2, "resp_ok", nil) @@ -983,6 +1030,96 @@ func TestOpenAIGatewayService_Forward_WSv2_HeaderSessionFallbackFromPromptCacheK require.True(t, gjson.Get(requestToJSONString(captureConn.lastWrite), "stream").Exists()) } +func TestOpenAIGatewayService_Forward_WSv2_CodexFingerprintHandshakeBodyParityAndDefaultCacheKey(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", nil) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set("originator", "codex_cli_rs") + c.Request.Header.Set("session-id", "header-session") + c.Request.Header.Set("x-codex-turn-metadata", `{"installation_id":"header-install","session_id":"header-session","thread_id":"header-thread","turn_id":"header-turn","window_id":"header-window","sandbox":"seatbelt"}`) + + cfg := &config.Config{} + cfg.Security.URLAllowlist.Enabled = false + cfg.Security.URLAllowlist.AllowInsecureHTTP = true + cfg.Gateway.OpenAIWS.Enabled = true + cfg.Gateway.OpenAIWS.OAuthEnabled = true + cfg.Gateway.OpenAIWS.APIKeyEnabled = true + cfg.Gateway.OpenAIWS.ResponsesWebsocketsV2 = true + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1 + + captureConn := &openAIWSCaptureConn{ + events: [][]byte{ + []byte(`{"type":"response.completed","response":{"id":"resp_ws_fingerprint","model":"gpt-5.2","usage":{"input_tokens":2,"output_tokens":1}}}`), + }, + } + captureDialer := &openAIWSCaptureDialer{conn: captureConn} + pool := newOpenAIWSConnPool(cfg) + pool.setClientDialerForTest(captureDialer) + + svc := &OpenAIGatewayService{ + cfg: cfg, + httpUpstream: &httpUpstreamRecorder{}, + cache: &stubGatewayCache{}, + openaiWSResolver: NewOpenAIWSProtocolResolver(cfg), + toolCorrector: NewCodexToolCorrector(), + openaiWSPool: pool, + } + account := newTestOAuthAccount(4405, map[string]any{ + codexFingerprintModeExtraKey: "session", + "responses_websockets_v2_enabled": true, + }) + account.Name = "oauth-ws-fingerprint" + account.Status = StatusActive + account.Schedulable = true + account.Concurrency = 1 + account.Credentials = map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"} + + body := []byte(`{"model":"gpt-5.2","stream":true,"prompt_cache_key":"body-session","client_metadata":{"session_id":"body-session","x-codex-turn-metadata":"{\"installation_id\":\"body-install\",\"session_id\":\"body-session\",\"thread_id\":\"body-thread\",\"turn_id\":\"body-turn\",\"window_id\":\"body-window\",\"sandbox\":\"seatbelt\"}"},"input":[{"type":"input_text","text":"hi"}]}`) + result, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, "resp_ws_fingerprint", result.RequestID) + require.NotNil(t, captureConn.lastWrite) + + seed, ok := codexFingerprintSeed(account.Extra) + require.True(t, ok) + wantInstall := resolveConvergedInstallationID(account, seed) + wantSession := resolveConvergedSessionID(seed) + wantThread := resolveConvergedThreadID(seed, "header-session") + payloadJSON := requestToJSONString(captureConn.lastWrite) + + require.Equal(t, wantInstall, captureDialer.lastHeaders.Get("x-codex-installation-id")) + require.Equal(t, wantSession, captureDialer.lastHeaders.Get("session-id")) + require.Equal(t, wantSession, captureDialer.lastHeaders.Get("session_id")) + require.Equal(t, wantThread, captureDialer.lastHeaders.Get("thread-id")) + require.Equal(t, wantThread, captureDialer.lastHeaders.Get("x-client-request-id")) + require.Equal(t, wantThread+":0", captureDialer.lastHeaders.Get("x-codex-window-id")) + + require.Equal(t, wantSession, gjson.Get(payloadJSON, "prompt_cache_key").String()) + require.Equal(t, wantInstall, gjson.Get(payloadJSON, "client_metadata.x-codex-installation-id").String()) + require.Equal(t, wantSession, gjson.Get(payloadJSON, "client_metadata.session_id").String()) + require.Equal(t, wantThread, gjson.Get(payloadJSON, "client_metadata.thread_id").String()) + require.Equal(t, wantThread+":0", gjson.Get(payloadJSON, "client_metadata.x-codex-window-id").String()) + + bodyTurnMetadata := gjson.Get(payloadJSON, "client_metadata.x-codex-turn-metadata").String() + headerTurnMetadata := captureDialer.lastHeaders.Get("x-codex-turn-metadata") + require.Equal(t, wantInstall, gjson.Get(bodyTurnMetadata, "installation_id").String()) + require.Equal(t, wantSession, gjson.Get(bodyTurnMetadata, "session_id").String()) + require.Equal(t, wantThread, gjson.Get(bodyTurnMetadata, "thread_id").String()) + require.Equal(t, wantSession, gjson.Get(headerTurnMetadata, "session_id").String()) + require.Equal(t, gjson.Get(bodyTurnMetadata, "turn_id").String(), gjson.Get(headerTurnMetadata, "turn_id").String()) + require.NotZero(t, gjson.Get(bodyTurnMetadata, "turn_started_at_unix_ms").Int()) + require.Equal(t, + gjson.Get(bodyTurnMetadata, "turn_started_at_unix_ms").Int(), + gjson.Get(headerTurnMetadata, "turn_started_at_unix_ms").Int(), + ) +} + func TestOpenAIGatewayService_Forward_WSv2_ResponseDoneUsageParsed(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index e0dbd58bd4..020d409d0b 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -62,6 +62,14 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( payload := s.buildOpenAIWSCreatePayload(reqBody, account) payloadStrategy, removedKeys := applyOpenAIWSRetryPayloadStrategy(payload, attempt) + turnState := "" + turnMetadata := "" + if c != nil && c.Request != nil { + turnState = strings.TrimSpace(c.GetHeader(openAIWSTurnStateHeader)) + turnMetadata = strings.TrimSpace(c.GetHeader(openAIWSTurnMetadataHeader)) + } + setOpenAIWSTurnMetadata(payload, turnMetadata) + applyStagedCodexFingerprintClientMetadata(c, account, payload) previousResponseID := openAIWSPayloadString(payload, "previous_response_id") previousResponseIDKind := ClassifyOpenAIPreviousResponseIDKind(previousResponseID) promptCacheKey := openAIWSPayloadString(payload, "prompt_cache_key") @@ -79,13 +87,6 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( if raw, ok := payload["stream"]; ok { streamValue = normalizeOpenAIWSLogValue(strings.TrimSpace(fmt.Sprintf("%v", raw))) } - turnState := "" - turnMetadata := "" - if c != nil && c.Request != nil { - turnState = strings.TrimSpace(c.GetHeader(openAIWSTurnStateHeader)) - turnMetadata = strings.TrimSpace(c.GetHeader(openAIWSTurnMetadataHeader)) - } - setOpenAIWSTurnMetadata(payload, turnMetadata) payloadEventType := openAIWSPayloadString(payload, "type") if payloadEventType == "" { payloadEventType = "response.create" diff --git a/backend/internal/service/openai_ws_pool.go b/backend/internal/service/openai_ws_pool.go index 048031d640..2c62b6b4c0 100644 --- a/backend/internal/service/openai_ws_pool.go +++ b/backend/internal/service/openai_ws_pool.go @@ -77,7 +77,13 @@ type openAIWSAcquireRequest struct { } type openAIWSHandshakeCompatibilityKey struct { - betaFeatures string + betaFeatures string + codexInstallationID string + sessionIDHyphen string + sessionIDUnderscore string + threadID string + clientRequestID string + codexWindowID string } type openAIWSConnLease struct { @@ -855,7 +861,7 @@ func (p *openAIWSConnPool) acquire(ctx context.Context, req openAIWSAcquireReque retryAcquire: accountID := req.Account.ID - compatibility := normalizeOpenAIWSHandshakeCompatibility(req.Headers) + compatibility := normalizeOpenAIWSHandshakeCompatibility(req.Account, req.Headers) routingAffinity := normalizeOpenAIWSRoutingAffinity(req.Headers) effectiveMaxConns := p.effectiveMaxConnsByAccount(req.Account) if effectiveMaxConns <= 0 { @@ -1015,37 +1021,39 @@ retryAcquire: p.ensureTargetIdleAsync(accountID) return lease, nil } - for _, conn := range ap.conns { - if conn == nil || conn == best || !conn.matchesHandshakeCompatibility(compatibility) || !conn.matchesRoutingAffinity(routingAffinity) { - continue - } - if conn.tryAcquire() { - connPick := time.Since(pickStartedAt) - p.recordConnPickDuration(connPick) - ap.mu.Unlock() - closeOpenAIWSConns(evicted) - if p.shouldHealthCheckConn(conn) { - if err := conn.pingWithTimeout(openAIWSConnHealthCheckTO); err != nil { - conn.close() - p.evictConn(accountID, conn.id) - if retry < 1 { - return p.acquire(ctx, req, retry+1) + if routingAffinity == "" || len(ap.conns)+ap.creating >= effectiveMaxConns { + for _, conn := range ap.conns { + if conn == nil || conn == best || !conn.matchesHandshakeCompatibility(compatibility) { + continue + } + if conn.tryAcquire() { + connPick := time.Since(pickStartedAt) + p.recordConnPickDuration(connPick) + ap.mu.Unlock() + closeOpenAIWSConns(evicted) + if p.shouldHealthCheckConn(conn) { + if err := conn.pingWithTimeout(openAIWSConnHealthCheckTO); err != nil { + conn.close() + p.evictConn(accountID, conn.id) + if retry < 1 { + return p.acquire(ctx, req, retry+1) + } + return nil, err } - return nil, err } + lease := &openAIWSConnLease{pool: p, accountID: accountID, conn: conn, connPick: connPick, reused: true} + p.metrics.acquireReuseTotal.Add(1) + p.recordLastSuccessfulAcquire(accountID, acquireGeneration, req) + p.ensureTargetIdleAsync(accountID) + return lease, nil } - lease := &openAIWSConnLease{pool: p, accountID: accountID, conn: conn, connPick: connPick, reused: true} - p.metrics.acquireReuseTotal.Add(1) - p.recordLastSuccessfulAcquire(accountID, acquireGeneration, req) - p.ensureTargetIdleAsync(accountID) - return lease, nil } } } if !req.ForceNewConn && len(ap.conns)+ap.creating >= effectiveMaxConns { affine := p.pickLeastBusyConnWithRoutingAffinityLocked(ap, compatibility, routingAffinity) - if idle := p.pickOldestIdleConnWithoutHandshakeCompatibilityOrRoutingAffinityLocked(ap, compatibility, routingAffinity); idle != nil { + if idle := p.pickOldestIdleConnWithoutHandshakeCompatibilityLocked(ap, compatibility); idle != nil { delete(ap.conns, idle.id) evicted = append(evicted, idle) p.metrics.scaleDownTotal.Add(1) @@ -1241,10 +1249,9 @@ func (p *openAIWSConnPool) pickOldestIdleConnLocked(ap *openAIWSAccountPool) *op return oldest } -func (p *openAIWSConnPool) pickOldestIdleConnWithoutHandshakeCompatibilityOrRoutingAffinityLocked( +func (p *openAIWSConnPool) pickOldestIdleConnWithoutHandshakeCompatibilityLocked( ap *openAIWSAccountPool, compatibility openAIWSHandshakeCompatibilityKey, - routingAffinity string, ) *openAIWSConn { if ap == nil || len(ap.conns) == 0 { return nil @@ -1252,7 +1259,7 @@ func (p *openAIWSConnPool) pickOldestIdleConnWithoutHandshakeCompatibilityOrRout var oldest *openAIWSConn for _, conn := range ap.conns { if conn == nil || - (conn.matchesHandshakeCompatibility(compatibility) && conn.matchesRoutingAffinity(routingAffinity)) || + conn.matchesHandshakeCompatibility(compatibility) || conn.isLeased() || conn.waiters.Load() > 0 || p.isConnPinnedLocked(ap, conn.id) { continue } @@ -1800,7 +1807,7 @@ func (p *openAIWSConnPool) dialConn(ctx context.Context, req openAIWSAcquireRequ } id := p.nextConnID(req.Account.ID) pooledConn := newOpenAIWSConn(id, req.Account.ID, conn, handshakeHeaders) - pooledConn.handshakeCompatibility = normalizeOpenAIWSHandshakeCompatibility(req.Headers) + pooledConn.handshakeCompatibility = normalizeOpenAIWSHandshakeCompatibility(req.Account, req.Headers) pooledConn.routingAffinity = normalizeOpenAIWSRoutingAffinity(req.Headers) return pooledConn, nil } @@ -1983,7 +1990,7 @@ func cloneOpenAIWSAcquireRequestPtr(req *openAIWSAcquireRequest) *openAIWSAcquir func sameOpenAIWSPrewarmTarget(a, b openAIWSAcquireRequest) bool { return stringsTrim(a.WSURL) == stringsTrim(b.WSURL) && stringsTrim(a.ProxyURL) == stringsTrim(b.ProxyURL) && - normalizeOpenAIWSHandshakeCompatibility(a.Headers) == normalizeOpenAIWSHandshakeCompatibility(b.Headers) + normalizeOpenAIWSHandshakeCompatibility(a.Account, a.Headers) == normalizeOpenAIWSHandshakeCompatibility(b.Account, b.Headers) } func normalizeOpenAIWSBetaFeatures(headers http.Header) string { @@ -2011,10 +2018,41 @@ func normalizeOpenAIWSBetaFeatures(headers http.Header) string { return strings.Join(normalized, ",") } -func normalizeOpenAIWSHandshakeCompatibility(headers http.Header) openAIWSHandshakeCompatibilityKey { - return openAIWSHandshakeCompatibilityKey{ +func normalizeOpenAIWSHandshakeCompatibility(account *Account, headers http.Header) openAIWSHandshakeCompatibilityKey { + key := openAIWSHandshakeCompatibilityKey{ betaFeatures: normalizeOpenAIWSBetaFeatures(headers), } + mode := activeCodexFingerprintMode(account) + if mode == codexFingerprintOff { + return key + } + key.codexInstallationID = normalizeOpenAIWSStableIdentityHeader(headers, "x-codex-installation-id") + if mode == codexFingerprintDevice { + return key + } + key.sessionIDHyphen = normalizeOpenAIWSStableIdentityHeader(headers, "session-id") + key.sessionIDUnderscore = normalizeOpenAIWSStableIdentityHeader(headers, "session_id") + key.threadID = normalizeOpenAIWSStableIdentityHeader(headers, "thread-id") + key.clientRequestID = normalizeOpenAIWSStableIdentityHeader(headers, "x-client-request-id") + key.codexWindowID = normalizeOpenAIWSStableIdentityHeader(headers, "x-codex-window-id") + return key +} + +func activeCodexFingerprintMode(account *Account) codexFingerprintMode { + if account == nil || account.GetCodexFingerprintMode() == codexFingerprintOff { + return codexFingerprintOff + } + if _, ok := codexFingerprintSeed(account.Extra); !ok { + return codexFingerprintOff + } + return account.GetCodexFingerprintMode() +} + +func normalizeOpenAIWSStableIdentityHeader(headers http.Header, name string) string { + if headers == nil { + return "" + } + return strings.TrimSpace(headers.Get(name)) } func normalizeOpenAIWSRoutingAffinity(headers http.Header) string { diff --git a/backend/internal/service/openai_ws_pool_test.go b/backend/internal/service/openai_ws_pool_test.go index f04b81d40d..ada18af8d3 100644 --- a/backend/internal/service/openai_ws_pool_test.go +++ b/backend/internal/service/openai_ws_pool_test.go @@ -623,6 +623,206 @@ func TestOpenAIWSConnPool_AcquireReusesOnlyMatchingBetaFeatures(t *testing.T) { require.Equal(t, 2, dialer.DialCount()) } +func activeCodexFingerprintPoolAccountForTest(id int64) *Account { + return &Account{ + ID: id, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Extra: map[string]any{ + codexFingerprintModeExtraKey: "session", + codexFingerprintSeedExtraKey: "11111111-1111-4111-8111-111111111111", + }, + } +} + +func stableOpenAIWSIdentityHeadersForTest() http.Header { + headers := make(http.Header) + headers.Set("X-Codex-Beta-Features", "remote_compaction_v2,responses_websockets_v2") + headers.Set("X-Codex-Installation-ID", "install-a") + headers.Set("session-id", "session-hyphen-a") + headers.Set("session_id", "session-underscore-a") + headers.Set("thread-id", "thread-a") + headers.Set("x-client-request-id", "client-request-a") + headers.Set("x-codex-window-id", "window-a") + return headers +} + +func TestOpenAIWSConnPool_AcquireReusesSameStableIdentityWithDifferentTurnMetadata(t *testing.T) { + cfg := &config.Config{} + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1 + + pool := newOpenAIWSConnPool(cfg) + dialer := &openAIWSCountingDialer{} + pool.setClientDialerForTest(dialer) + account := activeCodexFingerprintPoolAccountForTest(132) + headers := stableOpenAIWSIdentityHeadersForTest() + headers.Set("Authorization", "Bearer token-a") + headers.Set("x-codex-turn-metadata", `{"turn_id":"turn-a"}`) + + first, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: headers, + }) + require.NoError(t, err) + firstConnID := first.ConnID() + first.Release() + + nextHeaders := stableOpenAIWSIdentityHeadersForTest() + nextHeaders.Set("Authorization", "Bearer token-b") + nextHeaders.Set("x-codex-turn-metadata", `{"turn_id":"turn-b"}`) + nextHeaders.Set(openAICodexRoutingHintHeader, "model=gpt-5.6-codex;tier=priority") + second, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: nextHeaders, + }) + require.NoError(t, err) + require.True(t, second.Reused()) + require.Equal(t, firstConnID, second.ConnID()) + second.Release() + require.Equal(t, 1, dialer.DialCount(), "stable identity match should ignore auth, turn metadata, and soft routing hints") +} + +func TestOpenAIWSConnPool_AcquireDoesNotReuseDifferentStableIdentity(t *testing.T) { + for _, tt := range []struct { + name string + header string + value string + }{ + {name: "installation", header: "x-codex-installation-id", value: "install-b"}, + {name: "session hyphen", header: "session-id", value: "session-hyphen-b"}, + {name: "session underscore", header: "session_id", value: "session-underscore-b"}, + {name: "thread", header: "thread-id", value: "thread-b"}, + {name: "client request", header: "x-client-request-id", value: "client-request-b"}, + {name: "window", header: "x-codex-window-id", value: "window-b"}, + } { + t.Run(tt.name, func(t *testing.T) { + cfg := &config.Config{} + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 2 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 2 + + pool := newOpenAIWSConnPool(cfg) + dialer := &openAIWSCountingDialer{} + pool.setClientDialerForTest(dialer) + account := activeCodexFingerprintPoolAccountForTest(133) + + first, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: stableOpenAIWSIdentityHeadersForTest(), + }) + require.NoError(t, err) + firstConnID := first.ConnID() + first.Release() + + nextHeaders := stableOpenAIWSIdentityHeadersForTest() + nextHeaders.Set(tt.header, tt.value) + second, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: nextHeaders, + }) + require.NoError(t, err) + require.False(t, second.Reused()) + require.NotEqual(t, firstConnID, second.ConnID()) + second.Release() + require.Equal(t, 2, dialer.DialCount()) + }) + } +} + +func TestOpenAIWSConnPool_AcquireRoutingHintRemainsSoftAffinity(t *testing.T) { + cfg := &config.Config{} + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1 + + pool := newOpenAIWSConnPool(cfg) + dialer := &openAIWSCountingDialer{} + pool.setClientDialerForTest(dialer) + account := activeCodexFingerprintPoolAccountForTest(134) + + firstHeaders := stableOpenAIWSIdentityHeadersForTest() + firstHeaders.Set(openAICodexRoutingHintHeader, "model=gpt-5.6-codex") + first, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: firstHeaders, + }) + require.NoError(t, err) + firstConnID := first.ConnID() + first.Release() + + secondHeaders := stableOpenAIWSIdentityHeadersForTest() + secondHeaders.Set(openAICodexRoutingHintHeader, "model=gpt-5.6-codex;tier=priority") + second, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: secondHeaders, + }) + require.NoError(t, err) + require.True(t, second.Reused()) + require.Equal(t, firstConnID, second.ConnID()) + second.Release() + require.Equal(t, 1, dialer.DialCount()) +} + +func TestOpenAIWSConnPool_DeviceModeKeysOnlyInstallationIdentity(t *testing.T) { + cfg := &config.Config{} + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 2 + cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0 + cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 2 + + pool := newOpenAIWSConnPool(cfg) + dialer := &openAIWSCountingDialer{} + pool.setClientDialerForTest(dialer) + account := activeCodexFingerprintPoolAccountForTest(135) + account.Extra[codexFingerprintModeExtraKey] = "device" + + firstHeaders := stableOpenAIWSIdentityHeadersForTest() + first, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: firstHeaders, + }) + require.NoError(t, err) + firstConnID := first.ConnID() + first.Release() + + sessionChanged := stableOpenAIWSIdentityHeadersForTest() + sessionChanged.Set("session-id", "session-hyphen-b") + sessionChanged.Set("session_id", "session-underscore-b") + sessionChanged.Set("thread-id", "thread-b") + sessionChanged.Set("x-client-request-id", "client-request-b") + sessionChanged.Set("x-codex-window-id", "window-b") + second, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: sessionChanged, + }) + require.NoError(t, err) + require.True(t, second.Reused()) + require.Equal(t, firstConnID, second.ConnID()) + second.Release() + + installationChanged := sessionChanged.Clone() + installationChanged.Set("x-codex-installation-id", "install-b") + third, err := pool.Acquire(context.Background(), openAIWSAcquireRequest{ + Account: account, + WSURL: "wss://example.com/v1/responses", + Headers: installationChanged, + }) + require.NoError(t, err) + require.False(t, third.Reused()) + require.NotEqual(t, firstConnID, third.ConnID()) + third.Release() + require.Equal(t, 2, dialer.DialCount()) +} + func TestOpenAIWSConnPool_AcquireReplacesIdleConnWithDifferentBetaFeatures(t *testing.T) { cfg := &config.Config{} cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 diff --git a/backend/migrations/225_backfill_codex_fingerprint_seed.sql b/backend/migrations/225_backfill_codex_fingerprint_seed.sql new file mode 100644 index 0000000000..5c7bdbb2f6 --- /dev/null +++ b/backend/migrations/225_backfill_codex_fingerprint_seed.sql @@ -0,0 +1,21 @@ +-- Backfill system-managed Codex fingerprint seeds for enabled OpenAI OAuth accounts. +-- Idempotent: valid canonical seeds are preserved on rerun. +UPDATE accounts +SET extra = jsonb_set( + COALESCE(extra, '{}'::jsonb), + '{codex_fingerprint_seed}', + to_jsonb(gen_random_uuid()::text), + true +) +WHERE deleted_at IS NULL + AND platform = 'openai' + AND type = 'oauth' + AND COALESCE(extra->>'codex_fingerprint_mode', '') IN ('device', 'session', 'full') + AND ( + extra->>'codex_fingerprint_seed' IS NULL + OR btrim(extra->>'codex_fingerprint_seed') = '' + OR NOT ( + extra->>'codex_fingerprint_seed' ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' + AND extra->>'codex_fingerprint_seed' <> '00000000-0000-0000-0000-000000000000' + ) + );