Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[codex] add oauth pre-refresh resilience (#430)

Browse Source 
* oauth pre-refresh resilience

* address oauth refresh review comments
This commit is contained in:
Cita
2026-04-05 22:58:50 +08:00
committed by GitHub
parent 6feda0e7ba
commit 492f43c5dc
5 changed files with 1050 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/index.ts
+6
View File
@@ -45,6 +45,10 @@ import {
stopChannelRecoveryProbeScheduler,
} from './services/channelRecoveryProbeService.js';
import { startUpdateCenterPolling, stopUpdateCenterPolling } from './services/updateCenterPollingService.js';
import {
startOauthTokenRefreshScheduler,
stopOauthTokenRefreshScheduler,
} from './services/oauth/oauthRefreshScheduler.js';
import { reloadBackupWebdavScheduler } from './services/backupService.js';
import { ensureRuntimeDatabaseReady } from './runtimeDatabaseBootstrap.js';
import { isPublicApiRoute, registerDesktopRoutes } from './desktop.js';
@@ -255,6 +259,7 @@ startSiteAnnouncementPolling();
startModelAvailabilityProbeScheduler();
startChannelRecoveryProbeScheduler();
startUpdateCenterPolling();
startOauthTokenRefreshScheduler();
try {
await startOAuthLoopbackCallbackServers();
} catch (error) {
@@ -269,6 +274,7 @@ app.addHook('onClose', async () => {
stopProxyLogRetentionService();
stopModelAvailabilityProbeScheduler();
stopChannelRecoveryProbeScheduler();
await stopOauthTokenRefreshScheduler();
await stopOAuthLoopbackCallbackServers();
});
src/server/services/modelService.discovery.test.ts
+478
View File
@@ -949,6 +949,346 @@ describe('refreshModelsForAccount credential discovery', () => {
]);
});
it('refreshes claude oauth access token during cloud model discovery through singleflight and retries with the refreshed account state', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('claude oauth discovery should not call adapter.getModels'));
refreshOauthAccessTokenSingleflightMock.mockImplementation(async (accountId: number) => {
const refreshedExtraConfig = JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'claude',
email: 'claude-refreshed-user@example.com',
accountKey: 'claude-refreshed-user@example.com',
refreshToken: 'claude-refresh-token-next',
},
});
await db.update(schema.accounts).set({
accessToken: 'claude-access-token-refreshed',
oauthProvider: 'claude',
oauthAccountKey: 'claude-refreshed-user@example.com',
extraConfig: refreshedExtraConfig,
status: 'active',
updatedAt: '2026-03-21T00:00:00.000Z',
}).where(eq(schema.accounts.id, accountId)).run();
return {
accountId,
accessToken: 'claude-access-token-refreshed',
accountKey: 'claude-refreshed-user@example.com',
extraConfig: refreshedExtraConfig,
};
});
undiciFetchMock
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: true,
status: 200,
json: async () => ({
data: [
{ id: 'claude-sonnet-4-5-20250929' },
],
}),
text: async () => JSON.stringify({ ok: true }),
});
const site = await db.insert(schema.sites).values({
name: 'claude-refresh-site',
url: 'https://api.anthropic.com',
platform: 'claude',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'claude-user@example.com',
accessToken: 'claude-access-token-expired',
apiToken: null,
status: 'active',
oauthProvider: 'claude',
oauthAccountKey: 'claude-user@example.com',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'claude',
email: 'claude-user@example.com',
accountKey: 'claude-user@example.com',
refreshToken: 'claude-refresh-token',
},
}),
}).returning().get();
const result = await refreshModelsForAccount(account.id);
expect(result).toMatchObject({
accountId: account.id,
refreshed: true,
status: 'success',
modelCount: 1,
modelsPreview: ['claude-sonnet-4-5-20250929'],
discoveredByCredential: true,
});
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(account.id);
expect(undiciFetchMock).toHaveBeenCalledTimes(2);
expect(String(undiciFetchMock.mock.calls[0]?.[0] || '')).toBe('https://api.anthropic.com/v1/models');
expect(undiciFetchMock.mock.calls[0]?.[1]).toMatchObject({
method: 'GET',
headers: expect.objectContaining({
Authorization: 'Bearer claude-access-token-expired',
}),
});
expect(String(undiciFetchMock.mock.calls[1]?.[0] || '')).toBe('https://api.anthropic.com/v1/models');
expect(undiciFetchMock.mock.calls[1]?.[1]).toMatchObject({
method: 'GET',
headers: expect.objectContaining({
Authorization: 'Bearer claude-access-token-refreshed',
}),
});
const latest = await db.select().from(schema.accounts)
.where(eq(schema.accounts.id, account.id))
.get();
expect(latest).toMatchObject({
accessToken: 'claude-access-token-refreshed',
oauthProvider: 'claude',
oauthAccountKey: 'claude-refreshed-user@example.com',
});
const parsed = JSON.parse(latest!.extraConfig || '{}');
expect(parsed.oauth).toMatchObject({
email: 'claude-refreshed-user@example.com',
refreshToken: 'claude-refresh-token-next',
modelDiscoveryStatus: 'healthy',
});
expect(parsed.oauth).not.toHaveProperty('provider');
});
it('refreshes codex oauth access token during cloud model discovery through singleflight and retries with the refreshed account state', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('codex oauth discovery should not call adapter.getModels'));
refreshOauthAccessTokenSingleflightMock.mockImplementation(async (accountId: number) => {
const refreshedExtraConfig = JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
email: 'codex-refreshed-user@example.com',
accountId: 'chatgpt-account-refreshed',
planType: 'plus',
refreshToken: 'codex-refresh-token-next',
},
});
await db.update(schema.accounts).set({
accessToken: 'codex-access-token-refreshed',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-refreshed',
extraConfig: refreshedExtraConfig,
status: 'active',
updatedAt: '2026-03-21T00:00:00.000Z',
}).where(eq(schema.accounts.id, accountId)).run();
return {
accountId,
accessToken: 'codex-access-token-refreshed',
accountKey: 'chatgpt-account-refreshed',
extraConfig: refreshedExtraConfig,
};
});
undiciFetchMock
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: true,
status: 200,
json: async () => ({
models: [
{ id: 'gpt-5.4' },
],
}),
text: async () => JSON.stringify({ ok: true }),
});
const site = await db.insert(schema.sites).values({
name: 'codex-refresh-site',
url: 'https://chatgpt.com/backend-api/codex',
platform: 'codex',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'codex-user@example.com',
accessToken: 'codex-access-token-expired',
apiToken: null,
status: 'active',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-original',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
email: 'codex-user@example.com',
accountId: 'chatgpt-account-original',
planType: 'plus',
refreshToken: 'codex-refresh-token',
},
}),
}).returning().get();
const result = await refreshModelsForAccount(account.id);
expect(result).toMatchObject({
accountId: account.id,
refreshed: true,
status: 'success',
modelCount: 1,
modelsPreview: ['gpt-5.4'],
discoveredByCredential: true,
});
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(account.id);
expect(undiciFetchMock).toHaveBeenCalledTimes(2);
expect(String(undiciFetchMock.mock.calls[0]?.[0] || '')).toBe('https://chatgpt.com/backend-api/codex/models?client_version=1.0.0');
expect(undiciFetchMock.mock.calls[0]?.[1]).toMatchObject({
method: 'GET',
headers: expect.objectContaining({
Authorization: 'Bearer codex-access-token-expired',
}),
});
expect(String(undiciFetchMock.mock.calls[1]?.[0] || '')).toBe('https://chatgpt.com/backend-api/codex/models?client_version=1.0.0');
expect(undiciFetchMock.mock.calls[1]?.[1]).toMatchObject({
method: 'GET',
headers: expect.objectContaining({
Authorization: 'Bearer codex-access-token-refreshed',
}),
});
const latest = await db.select().from(schema.accounts)
.where(eq(schema.accounts.id, account.id))
.get();
expect(latest).toMatchObject({
accessToken: 'codex-access-token-refreshed',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-refreshed',
});
const parsed = JSON.parse(latest!.extraConfig || '{}');
expect(parsed.oauth).toMatchObject({
email: 'codex-refreshed-user@example.com',
refreshToken: 'codex-refresh-token-next',
modelDiscoveryStatus: 'healthy',
});
expect(parsed.oauth).not.toHaveProperty('provider');
});
it('preserves refreshed codex oauth metadata when the retry after refresh still fails', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('codex oauth discovery should not call adapter.getModels'));
refreshOauthAccessTokenSingleflightMock.mockImplementation(async (accountId: number) => {
const refreshedExtraConfig = JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
email: 'codex-refreshed-user@example.com',
accountId: 'chatgpt-account-refreshed',
planType: 'plus',
refreshToken: 'codex-refresh-token-next',
tokenExpiresAt: 1770000000000,
},
});
await db.update(schema.accounts).set({
accessToken: 'codex-access-token-refreshed',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-refreshed',
extraConfig: refreshedExtraConfig,
status: 'active',
updatedAt: '2026-03-21T00:00:00.000Z',
}).where(eq(schema.accounts.id, accountId)).run();
return {
accountId,
accessToken: 'codex-access-token-refreshed',
accountKey: 'chatgpt-account-refreshed',
extraConfig: refreshedExtraConfig,
};
});
undiciFetchMock
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: false,
status: 503,
json: async () => ({ error: 'still unavailable' }),
text: async () => 'still unavailable',
});
const site = await db.insert(schema.sites).values({
name: 'codex-refresh-failure-site',
url: 'https://chatgpt.com/backend-api/codex',
platform: 'codex',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'codex-user@example.com',
accessToken: 'codex-access-token-expired',
apiToken: null,
status: 'active',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-original',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
email: 'codex-user@example.com',
accountId: 'chatgpt-account-original',
planType: 'plus',
refreshToken: 'codex-refresh-token-old',
tokenExpiresAt: 1760000000000,
},
}),
}).returning().get();
const result = await refreshModelsForAccount(account.id);
expect(result).toMatchObject({
accountId: account.id,
refreshed: true,
status: 'failed',
errorCode: 'unknown',
discoveredByCredential: false,
});
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(account.id);
const latest = await db.select().from(schema.accounts)
.where(eq(schema.accounts.id, account.id))
.get();
expect(latest).toMatchObject({
accessToken: 'codex-access-token-refreshed',
oauthProvider: 'codex',
oauthAccountKey: 'chatgpt-account-refreshed',
});
const parsed = JSON.parse(latest!.extraConfig || '{}');
expect(parsed.oauth).toMatchObject({
email: 'codex-refreshed-user@example.com',
refreshToken: 'codex-refresh-token-next',
tokenExpiresAt: 1770000000000,
modelDiscoveryStatus: 'abnormal',
});
});
it('marks codex oauth account abnormal when upstream cloud discovery fails', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('codex plan discovery should not call adapter.getModels'));
@@ -1311,6 +1651,144 @@ describe('refreshModelsForAccount credential discovery', () => {
expect(parsed.oauth).not.toHaveProperty('projectId');
});
it('refreshes antigravity oauth access token during model discovery through singleflight and retries with the refreshed account state', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('antigravity oauth discovery should not call adapter.getModels'));
refreshOauthAccessTokenSingleflightMock.mockImplementation(async (accountId: number) => {
const refreshedExtraConfig = JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'antigravity',
email: 'antigravity-refreshed-user@example.com',
accountId: 'antigravity-refreshed-user@example.com',
accountKey: 'antigravity-refreshed-user@example.com',
projectId: 'project-refresh-demo',
refreshToken: 'antigravity-refresh-token-next',
},
});
await db.update(schema.accounts).set({
accessToken: 'antigravity-access-token-refreshed',
oauthProvider: 'antigravity',
oauthAccountKey: 'antigravity-refreshed-user@example.com',
oauthProjectId: 'project-refresh-demo',
extraConfig: refreshedExtraConfig,
status: 'active',
updatedAt: '2026-03-21T00:00:00.000Z',
}).where(eq(schema.accounts.id, accountId)).run();
return {
accountId,
accessToken: 'antigravity-access-token-refreshed',
accountKey: 'antigravity-refreshed-user@example.com',
extraConfig: refreshedExtraConfig,
};
});
undiciFetchMock
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: false,
status: 401,
json: async () => ({ error: 'expired' }),
text: async () => 'expired',
})
.mockResolvedValueOnce({
ok: true,
status: 200,
json: async () => ({
models: {
'gemini-3-pro-preview': { displayName: 'Gemini 3 Pro Preview' },
},
}),
text: async () => JSON.stringify({ ok: true }),
});
const site = await db.insert(schema.sites).values({
name: 'antigravity-refresh-site',
url: 'https://cloudcode-pa.googleapis.com',
platform: 'antigravity',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'antigravity-user@example.com',
accessToken: 'antigravity-access-token-expired',
apiToken: null,
status: 'active',
oauthProvider: 'antigravity',
oauthAccountKey: 'antigravity-user@example.com',
oauthProjectId: 'project-refresh-demo',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'antigravity',
email: 'antigravity-user@example.com',
accountId: 'antigravity-user@example.com',
accountKey: 'antigravity-user@example.com',
projectId: 'project-refresh-demo',
refreshToken: 'antigravity-refresh-token',
},
}),
}).returning().get();
const result = await refreshModelsForAccount(account.id);
expect(result).toMatchObject({
accountId: account.id,
refreshed: true,
status: 'success',
modelCount: 1,
modelsPreview: ['gemini-3-pro-preview'],
discoveredByCredential: true,
});
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(account.id);
expect(undiciFetchMock).toHaveBeenCalledTimes(4);
expect(String(undiciFetchMock.mock.calls[0]?.[0] || '')).toBe('https://cloudcode-pa.googleapis.com/v1internal:fetchAvailableModels');
expect(undiciFetchMock.mock.calls[0]?.[1]).toMatchObject({
method: 'POST',
headers: expect.objectContaining({
Authorization: 'Bearer antigravity-access-token-expired',
}),
});
expect(String(undiciFetchMock.mock.calls[3]?.[0] || '')).toBe('https://cloudcode-pa.googleapis.com/v1internal:fetchAvailableModels');
expect(undiciFetchMock.mock.calls[3]?.[1]).toMatchObject({
method: 'POST',
headers: expect.objectContaining({
Authorization: 'Bearer antigravity-access-token-refreshed',
}),
});
const latest = await db.select().from(schema.accounts)
.where(eq(schema.accounts.id, account.id))
.get();
expect(latest).toMatchObject({
accessToken: 'antigravity-access-token-refreshed',
oauthProvider: 'antigravity',
oauthAccountKey: 'antigravity-refreshed-user@example.com',
oauthProjectId: 'project-refresh-demo',
});
const parsed = JSON.parse(latest!.extraConfig || '{}');
expect(parsed.oauth).toMatchObject({
email: 'antigravity-refreshed-user@example.com',
refreshToken: 'antigravity-refresh-token-next',
modelDiscoveryStatus: 'healthy',
});
expect(parsed.oauth).not.toHaveProperty('provider');
expect(parsed.oauth).not.toHaveProperty('projectId');
});
it('discovers antigravity oauth models via fetchAvailableModels fallback using the oauth project id', async () => {
getApiTokenMock.mockResolvedValue(null);
getModelsMock.mockRejectedValue(new Error('antigravity oauth discovery should not call adapter.getModels'));
src/server/services/modelService.ts
+86 -15
View File
@@ -108,6 +108,11 @@ export type ModelRefreshResult =
| ModelRefreshFailureResult
| ModelRefreshSuccessResult;
type ModelDiscoveryAccountRow = typeof schema.accounts.$inferSelect;
type ModelDiscoveryRetryError = {
refreshedAccount?: ModelDiscoveryAccountRow;
};
function looksLikeHtmlJsonParseError(message: string): boolean {
const lowered = String(message || '').trim().toLowerCase();
return (
@@ -303,6 +308,54 @@ function shouldRetryModelDiscoveryWithOauthRefresh(error: unknown): boolean {
|| message.includes('unauthenticated');
}
async function retryOauthModelDiscoveryWithRefresh<T>(input: {
account: ModelDiscoveryAccountRow;
attempt: (account: ModelDiscoveryAccountRow) => Promise<T>;
}): Promise<{ result: T; account: ModelDiscoveryAccountRow }> {
let discoveryAccount = input.account;
try {
return {
result: await input.attempt(discoveryAccount),
account: discoveryAccount,
};
} catch (error) {
if (!shouldRetryModelDiscoveryWithOauthRefresh(error)) {
throw error;
}
await refreshOauthAccessTokenSingleflight(discoveryAccount.id);
const refreshedAccount = await db.select().from(schema.accounts)
.where(eq(schema.accounts.id, discoveryAccount.id))
.get();
if (!refreshedAccount) {
throw error;
}
discoveryAccount = refreshedAccount;
try {
return {
result: await input.attempt(discoveryAccount),
account: discoveryAccount,
};
} catch (retryError) {
if (retryError && typeof retryError === 'object') {
(retryError as ModelDiscoveryRetryError).refreshedAccount = discoveryAccount;
throw retryError;
}
const wrappedError = new Error(String(retryError || 'model discovery retry failed')) as Error & ModelDiscoveryRetryError;
wrappedError.refreshedAccount = discoveryAccount;
throw wrappedError;
}
}
}
function extractRefreshedDiscoveryAccount(error: unknown): ModelDiscoveryAccountRow | null {
if (!error || typeof error !== 'object') return null;
const refreshedAccount = (error as ModelDiscoveryRetryError).refreshedAccount;
return refreshedAccount ?? null;
}
export async function refreshModelsForAccount(
accountId: number,
options?: { allowInactive?: boolean },
@@ -405,13 +458,18 @@ export async function refreshModelsForAccount(
if (oauth?.provider === 'codex') {
const checkedAt = new Date().toISOString();
const startedAt = Date.now();
let discoveryAccount = account;
try {
const codexModels = await withTimeout(
const { result: codexModels, account: refreshedAccount } = await retryOauthModelDiscoveryWithRefresh({
account,
attempt: async (candidateAccount) => withTimeout(
() => withAccountProxyOverride(accountProxyUrl,
() => discoverCodexModelsFromCloud({ site, account })),
() => discoverCodexModelsFromCloud({ site, account: candidateAccount })),
MODEL_DISCOVERY_TIMEOUT_MS,
`codex model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`,
);
),
});
discoveryAccount = refreshedAccount;
if (codexModels.length === 0) {
throw new Error('未获取到可用模型');
}
@@ -429,7 +487,7 @@ export async function refreshModelsForAccount(
).run();
}
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'healthy',
lastDiscoveredModels: codexModels,
@@ -449,11 +507,12 @@ export async function refreshModelsForAccount(
discoveredApiToken: false,
});
} catch (err) {
discoveryAccount = extractRefreshedDiscoveryAccount(err) ?? discoveryAccount;
const rawMessage = (err as { message?: string })?.message || 'codex model discovery failed';
const errorCode = classifyModelDiscoveryError(rawMessage);
const errorMessage = `Codex 模型获取失败（${rawMessage}）`;
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'abnormal',
lastModelSyncError: errorMessage,
@@ -480,13 +539,18 @@ export async function refreshModelsForAccount(
if (oauth?.provider === 'claude') {
const checkedAt = new Date().toISOString();
const startedAt = Date.now();
let discoveryAccount = account;
try {
const claudeModels = await withTimeout(
const { result: claudeModels, account: refreshedAccount } = await retryOauthModelDiscoveryWithRefresh({
account,
attempt: async (candidateAccount) => withTimeout(
() => withAccountProxyOverride(accountProxyUrl,
() => discoverClaudeModelsFromCloud({ site, account })),
() => discoverClaudeModelsFromCloud({ site, account: candidateAccount })),
MODEL_DISCOVERY_TIMEOUT_MS,
`claude oauth model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`,
);
),
});
discoveryAccount = refreshedAccount;
if (claudeModels.length === 0) {
throw new Error('未获取到可用模型');
}
@@ -503,7 +567,7 @@ export async function refreshModelsForAccount(
).run();
}
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'healthy',
lastDiscoveredModels: claudeModels,
@@ -523,11 +587,12 @@ export async function refreshModelsForAccount(
discoveredApiToken: false,
});
} catch (err) {
discoveryAccount = extractRefreshedDiscoveryAccount(err) ?? discoveryAccount;
const rawMessage = (err as { message?: string })?.message || 'claude oauth model discovery failed';
const errorCode = classifyModelDiscoveryError(rawMessage);
const errorMessage = `Claude OAuth 模型获取失败（${rawMessage}）`;
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'abnormal',
lastModelSyncError: errorMessage,
@@ -647,13 +712,18 @@ export async function refreshModelsForAccount(
if (oauth?.provider === 'antigravity') {
const checkedAt = new Date().toISOString();
const startedAt = Date.now();
let discoveryAccount = account;
try {
const antigravityModels = await withTimeout(
const { result: antigravityModels, account: refreshedAccount } = await retryOauthModelDiscoveryWithRefresh({
account,
attempt: async (candidateAccount) => withTimeout(
() => withAccountProxyOverride(accountProxyUrl,
() => discoverAntigravityModelsFromCloud({ site, account })),
() => discoverAntigravityModelsFromCloud({ site, account: candidateAccount })),
MODEL_DISCOVERY_TIMEOUT_MS,
`antigravity model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`,
);
),
});
discoveryAccount = refreshedAccount;
if (antigravityModels.length === 0) {
throw new Error('未获取到可用模型');
}
@@ -671,7 +741,7 @@ export async function refreshModelsForAccount(
).run();
}
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'healthy',
lastDiscoveredModels: antigravityModels,
@@ -691,11 +761,12 @@ export async function refreshModelsForAccount(
discoveredApiToken: false,
});
} catch (err) {
discoveryAccount = extractRefreshedDiscoveryAccount(err) ?? discoveryAccount;
const rawMessage = (err as { message?: string })?.message || 'antigravity model discovery failed';
const errorCode = classifyModelDiscoveryError(rawMessage);
const errorMessage = `Antigravity 模型获取失败（${rawMessage}）`;
await updateOauthModelDiscoveryState({
account,
account: discoveryAccount,
checkedAt,
status: 'abnormal',
lastModelSyncError: errorMessage,
src/server/services/oauth/oauthRefreshScheduler.test.ts
+331
View File
@@ -0,0 +1,331 @@
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
const refreshOauthAccessTokenSingleflightMock = vi.fn();
vi.mock('./refreshSingleflight.js', () => ({
refreshOauthAccessTokenSingleflight: (...args: unknown[]) => refreshOauthAccessTokenSingleflightMock(...args),
}));
type DbModule = typeof import('../../db/index.js');
type SchedulerModule = typeof import('./oauthRefreshScheduler.js');
function buildOauthExtraConfig(input: {
provider: string;
refreshToken?: string;
tokenExpiresAt?: number;
}): string {
return JSON.stringify({
credentialMode: 'session',
oauth: {
provider: input.provider,
email: `${input.provider}-user@example.com`,
accountKey: `${input.provider}-user@example.com`,
...(input.refreshToken ? { refreshToken: input.refreshToken } : {}),
...(input.tokenExpiresAt ? { tokenExpiresAt: input.tokenExpiresAt } : {}),
},
});
}
describe('oauthRefreshScheduler', () => {
let db: DbModule['db'];
let schema: DbModule['schema'];
let executeOauthTokenAutoRefreshPass: SchedulerModule['executeOauthTokenAutoRefreshPass'];
let startOauthTokenRefreshScheduler: SchedulerModule['startOauthTokenRefreshScheduler'];
let stopOauthTokenRefreshScheduler: SchedulerModule['stopOauthTokenRefreshScheduler'];
let resetOauthTokenRefreshSchedulerForTests: SchedulerModule['__resetOauthTokenRefreshSchedulerForTests'];
let dataDir = '';
let originalDataDir: string | undefined;
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-oauth-refresh-scheduler-'));
originalDataDir = process.env.DATA_DIR;
process.env.DATA_DIR = dataDir;
await import('../../db/migrate.js');
const dbModule = await import('../../db/index.js');
const schedulerModule = await import('./oauthRefreshScheduler.js');
db = dbModule.db;
schema = dbModule.schema;
executeOauthTokenAutoRefreshPass = schedulerModule.executeOauthTokenAutoRefreshPass;
startOauthTokenRefreshScheduler = schedulerModule.startOauthTokenRefreshScheduler;
stopOauthTokenRefreshScheduler = schedulerModule.stopOauthTokenRefreshScheduler;
resetOauthTokenRefreshSchedulerForTests = schedulerModule.__resetOauthTokenRefreshSchedulerForTests;
});
beforeEach(async () => {
vi.useFakeTimers();
refreshOauthAccessTokenSingleflightMock.mockReset();
await stopOauthTokenRefreshScheduler();
await resetOauthTokenRefreshSchedulerForTests();
await db.delete(schema.accounts).run();
await db.delete(schema.sites).run();
});
afterEach(() => {
return (async () => {
await stopOauthTokenRefreshScheduler();
await resetOauthTokenRefreshSchedulerForTests();
vi.useRealTimers();
})();
});
afterAll(() => {
if (originalDataDir === undefined) {
delete process.env.DATA_DIR;
} else {
process.env.DATA_DIR = originalDataDir;
}
});
it('refreshes only oauth accounts that are within the provider-specific lead window, including gemini-cli', async () => {
const nowMs = Date.parse('2026-04-05T12:00:00.000Z');
vi.setSystemTime(nowMs);
const activeSite = await db.insert(schema.sites).values({
name: 'oauth-refresh-site',
url: 'https://oauth-refresh.example.com',
platform: 'codex',
status: 'active',
}).returning().get();
const disabledSite = await db.insert(schema.sites).values({
name: 'oauth-disabled-site',
url: 'https://oauth-disabled.example.com',
platform: 'codex',
status: 'disabled',
}).returning().get();
const accountIds: Record<string, number> = {};
for (const account of [
{
key: 'codex_due',
provider: 'codex',
siteId: activeSite.id,
status: 'active',
refreshToken: 'codex-refresh',
tokenExpiresAt: nowMs + (4 * 24 * 60 * 60 * 1000),
},
{
key: 'codex_skip',
provider: 'codex',
siteId: activeSite.id,
status: 'active',
refreshToken: 'codex-refresh-later',
tokenExpiresAt: nowMs + (6 * 24 * 60 * 60 * 1000),
},
{
key: 'claude_due',
provider: 'claude',
siteId: activeSite.id,
status: 'active',
refreshToken: 'claude-refresh',
tokenExpiresAt: nowMs + (3 * 60 * 60 * 1000),
},
{
key: 'claude_skip',
provider: 'claude',
siteId: activeSite.id,
status: 'active',
refreshToken: 'claude-refresh-later',
tokenExpiresAt: nowMs + (5 * 60 * 60 * 1000),
},
{
key: 'antigravity_due',
provider: 'antigravity',
siteId: activeSite.id,
status: 'active',
refreshToken: 'antigravity-refresh',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
},
{
key: 'antigravity_skip',
provider: 'antigravity',
siteId: activeSite.id,
status: 'active',
refreshToken: 'antigravity-refresh-later',
tokenExpiresAt: nowMs + (6 * 60 * 1000),
},
{
key: 'gemini_due',
provider: 'gemini-cli',
siteId: activeSite.id,
status: 'active',
refreshToken: 'gemini-refresh',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
},
{
key: 'gemini_skip',
provider: 'gemini-cli',
siteId: activeSite.id,
status: 'active',
refreshToken: 'gemini-refresh-later',
tokenExpiresAt: nowMs + (6 * 60 * 1000),
},
{
key: 'inactive_due',
provider: 'antigravity',
siteId: activeSite.id,
status: 'disabled',
refreshToken: 'inactive-refresh',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
},
{
key: 'disabled_site_due',
provider: 'antigravity',
siteId: disabledSite.id,
status: 'active',
refreshToken: 'site-disabled-refresh',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
},
{
key: 'missing_refresh_token',
provider: 'antigravity',
siteId: activeSite.id,
status: 'active',
refreshToken: '',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
},
{
key: 'missing_expiry',
provider: 'antigravity',
siteId: activeSite.id,
status: 'active',
refreshToken: 'missing-expiry-refresh',
},
]) {
const inserted = await db.insert(schema.accounts).values({
siteId: account.siteId,
username: `${account.key}@example.com`,
accessToken: `${account.key}-access-token`,
apiToken: null,
status: account.status,
oauthProvider: account.provider,
oauthAccountKey: `${account.key}@example.com`,
extraConfig: buildOauthExtraConfig({
provider: account.provider,
refreshToken: account.refreshToken,
tokenExpiresAt: account.tokenExpiresAt,
}),
}).returning().get();
accountIds[account.key] = inserted.id;
}
const result = await executeOauthTokenAutoRefreshPass({ nowMs });
expect(result).toMatchObject({
scanned: 12,
refreshed: 4,
failed: 0,
skipped: 8,
});
expect(result.refreshedAccountIds.sort((a, b) => a - b)).toEqual([
accountIds.codex_due,
accountIds.claude_due,
accountIds.antigravity_due,
accountIds.gemini_due,
].sort((a, b) => a - b));
expect(result.failedAccountIds).toEqual([]);
expect(refreshOauthAccessTokenSingleflightMock.mock.calls.map((call) => call[0]).sort((a, b) => a - b)).toEqual([
accountIds.codex_due,
accountIds.claude_due,
accountIds.antigravity_due,
accountIds.gemini_due,
].sort((a, b) => a - b));
});
it('starts with an immediate pass and keeps polling until stopped', async () => {
const nowMs = Date.parse('2026-04-05T12:00:00.000Z');
vi.setSystemTime(nowMs);
const site = await db.insert(schema.sites).values({
name: 'oauth-refresh-site',
url: 'https://oauth-refresh.example.com',
platform: 'antigravity',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'scheduler-user@example.com',
accessToken: 'scheduler-access-token',
apiToken: null,
status: 'active',
oauthProvider: 'antigravity',
oauthAccountKey: 'scheduler-user@example.com',
extraConfig: buildOauthExtraConfig({
provider: 'antigravity',
refreshToken: 'scheduler-refresh-token',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
}),
}).returning().get();
const started = startOauthTokenRefreshScheduler(60_000);
expect(started).toMatchObject({
enabled: true,
intervalMs: 60_000,
});
await vi.advanceTimersByTimeAsync(0);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledTimes(1);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenLastCalledWith(account.id);
await vi.advanceTimersByTimeAsync(60_000);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledTimes(2);
await stopOauthTokenRefreshScheduler();
await vi.advanceTimersByTimeAsync(60_000);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledTimes(2);
});
it('waits for an in-flight refresh pass before stop resolves', async () => {
const nowMs = Date.parse('2026-04-05T12:00:00.000Z');
vi.setSystemTime(nowMs);
let releaseRefresh: (() => void) | null = null;
refreshOauthAccessTokenSingleflightMock.mockImplementation(() => new Promise<void>((resolve) => {
releaseRefresh = resolve;
}));
const site = await db.insert(schema.sites).values({
name: 'oauth-refresh-site',
url: 'https://oauth-refresh.example.com',
platform: 'antigravity',
status: 'active',
}).returning().get();
await db.insert(schema.accounts).values({
siteId: site.id,
username: 'scheduler-user@example.com',
accessToken: 'scheduler-access-token',
apiToken: null,
status: 'active',
oauthProvider: 'antigravity',
oauthAccountKey: 'scheduler-user@example.com',
extraConfig: buildOauthExtraConfig({
provider: 'antigravity',
refreshToken: 'scheduler-refresh-token',
tokenExpiresAt: nowMs + (4 * 60 * 1000),
}),
}).run();
startOauthTokenRefreshScheduler(60_000);
await vi.advanceTimersByTimeAsync(0);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledTimes(1);
let stopResolved = false;
const stopPromise = Promise.resolve(stopOauthTokenRefreshScheduler()).then(() => {
stopResolved = true;
});
await Promise.resolve();
expect(stopResolved).toBe(false);
releaseRefresh?.();
await stopPromise;
expect(stopResolved).toBe(true);
});
});
src/server/services/oauth/oauthRefreshScheduler.ts
+140
View File
@@ -0,0 +1,140 @@
import { eq } from 'drizzle-orm';
import { db, schema } from '../../db/index.js';
import { getOauthInfoFromAccount } from './oauthAccount.js';
import { refreshOauthAccessTokenSingleflight } from './refreshSingleflight.js';
const OAUTH_REFRESH_SCHEDULER_INTERVAL_MS = 60_000;
const DEFAULT_OAUTH_REFRESH_LEAD_MS = 5 * 60 * 1000;
const OAUTH_REFRESH_LEAD_BY_PROVIDER: Record<string, number> = {
codex: 5 * 24 * 60 * 60 * 1000,
claude: 4 * 60 * 60 * 1000,
'gemini-cli': 5 * 60 * 1000,
antigravity: 5 * 60 * 1000,
};
let oauthRefreshSchedulerTimer: ReturnType<typeof setInterval> | null = null;
let oauthRefreshPassInFlight: Promise<void> | null = null;
function clearOauthRefreshSchedulerTimer(): void {
if (!oauthRefreshSchedulerTimer) return;
clearInterval(oauthRefreshSchedulerTimer);
oauthRefreshSchedulerTimer = null;
}
function normalizeProvider(provider?: string | null): string {
return String(provider || '').trim().toLowerCase();
}
function getOauthRefreshLeadMs(provider?: string | null): number {
const normalizedProvider = normalizeProvider(provider);
return OAUTH_REFRESH_LEAD_BY_PROVIDER[normalizedProvider] ?? DEFAULT_OAUTH_REFRESH_LEAD_MS;
}
function shouldRefreshOauthAccount(input: {
account: typeof schema.accounts.$inferSelect;
site: typeof schema.sites.$inferSelect;
nowMs: number;
}): boolean {
if ((input.account.status || 'active') !== 'active') return false;
if ((input.site.status || 'active') !== 'active') return false;
const oauth = getOauthInfoFromAccount(input.account);
if (!oauth?.refreshToken) return false;
if (!(typeof oauth.tokenExpiresAt === 'number' && Number.isFinite(oauth.tokenExpiresAt) && oauth.tokenExpiresAt > 0)) {
return false;
}
return oauth.tokenExpiresAt - input.nowMs <= getOauthRefreshLeadMs(oauth.provider);
}
export async function executeOauthTokenAutoRefreshPass(input: {
nowMs?: number;
} = {}) {
const nowMs = typeof input.nowMs === 'number' && Number.isFinite(input.nowMs)
? input.nowMs
: Date.now();
const rows = await db.select()
.from(schema.accounts)
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
.all();
const oauthRows = rows.filter((row) => !!getOauthInfoFromAccount(row.accounts));
const refreshedAccountIds: number[] = [];
const failedAccountIds: number[] = [];
let skipped = 0;
for (const row of oauthRows) {
if (!shouldRefreshOauthAccount({
account: row.accounts,
site: row.sites,
nowMs,
})) {
skipped += 1;
continue;
}
try {
await refreshOauthAccessTokenSingleflight(row.accounts.id);
refreshedAccountIds.push(row.accounts.id);
} catch (error) {
failedAccountIds.push(row.accounts.id);
console.warn(
`[oauth-refresh] failed to refresh account ${row.accounts.id}: ${(error as Error)?.message || 'unknown error'}`,
);
}
}
return {
scanned: oauthRows.length,
refreshed: refreshedAccountIds.length,
failed: failedAccountIds.length,
skipped,
refreshedAccountIds,
failedAccountIds,
};
}
async function runScheduledOauthRefreshPass(): Promise<void> {
if (oauthRefreshPassInFlight) {
return oauthRefreshPassInFlight;
}
oauthRefreshPassInFlight = executeOauthTokenAutoRefreshPass()
.then(() => undefined)
.catch((error) => {
console.warn(`[oauth-refresh] scheduled pass failed: ${(error as Error)?.message || 'unknown error'}`);
})
.finally(() => {
oauthRefreshPassInFlight = null;
});
return oauthRefreshPassInFlight;
}
export function startOauthTokenRefreshScheduler(intervalMs = OAUTH_REFRESH_SCHEDULER_INTERVAL_MS) {
clearOauthRefreshSchedulerTimer();
const safeIntervalMs = Math.max(OAUTH_REFRESH_SCHEDULER_INTERVAL_MS, Math.trunc(intervalMs || 0));
void runScheduledOauthRefreshPass();
oauthRefreshSchedulerTimer = setInterval(() => {
void runScheduledOauthRefreshPass();
}, safeIntervalMs);
oauthRefreshSchedulerTimer.unref?.();
return {
enabled: true,
intervalMs: safeIntervalMs,
};
}
export async function stopOauthTokenRefreshScheduler() {
clearOauthRefreshSchedulerTimer();
if (oauthRefreshPassInFlight) {
await oauthRefreshPassInFlight;
}
}
export async function __resetOauthTokenRefreshSchedulerForTests() {
await stopOauthTokenRefreshScheduler();
oauthRefreshPassInFlight = null;
}