diff --git a/src/providers/claude/claude-kiro.js b/src/providers/claude/claude-kiro.js index 77feea4..e0d0434 100644 --- a/src/providers/claude/claude-kiro.js +++ b/src/providers/claude/claude-kiro.js @@ -45,9 +45,9 @@ const KIRO_MODELS = getProviderModels('claude-kiro-oauth'); // 完整的模型映射表 const FULL_MODEL_MAPPING = { + "claude-haiku-4-5":"claude-haiku-4.5", "claude-opus-4-5":"claude-opus-4.5", "claude-opus-4-5-20251101":"claude-opus-4.5", - "claude-haiku-4-5":"claude-haiku-4.5", "claude-sonnet-4-5": "CLAUDE_SONNET_4_5_20250929_V1_0", "claude-sonnet-4-5-20250929": "CLAUDE_SONNET_4_5_20250929_V1_0", "claude-sonnet-4-20250514": "CLAUDE_SONNET_4_20250514_V1_0", @@ -1339,7 +1339,7 @@ async saveCredentialsToFile(filePath, newData) { } // 标记当前凭证为不健康（会自动进入刷新队列） - this._markCredentialUnhealthy('401 Unauthorized - Triggering auto-refresh'); + this._markCredentialNeedRefresh('401 Unauthorized - Triggering auto-refresh'); // Mark error for credential switch without recording error count error.shouldSwitchCredential = true; error.skipErrorCount = true; @@ -1354,7 +1354,7 @@ async saveCredentialsToFile(filePath, newData) { // Handle 403 (Forbidden) - mark as unhealthy immediately, no retry if (status === 403) { console.log('[Kiro] Received 403. Marking credential as unhealthy...'); - // this._markCredentialUnhealthy('403 Forbidden', error); + this._markCredentialUnhealthy('403 Forbidden', error); // Mark error for credential switch without recording error count error.shouldSwitchCredential = true; error.skipErrorCount = true; @@ -1421,7 +1421,7 @@ async saveCredentialsToFile(filePath, newData) { * @returns {boolean} - Whether the credential was successfully marked as unhealthy * @private */ - _markCredentialUnhealthy(reason, error = null) { + _markCredentialNeedRefresh(reason, error = null) { const poolManager = getProviderPoolManager(); if (poolManager && this.uuid) { console.log(`[Kiro] Marking credential ${this.uuid} as needs refresh. Reason: ${reason}`); @@ -1439,6 +1439,31 @@ async saveCredentialsToFile(filePath, newData) { return false; } } + + /** + * Helper method to mark the current credential as unhealthy + * @param {string} reason - The reason for marking unhealthy + * @param {Error} [error] - Optional error object to attach the marker to + * @returns {boolean} - Whether the credential was successfully marked as unhealthy + * @private + */ + _markCredentialUnhealthy(reason, error = null) { + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Kiro] Marking credential ${this.uuid} as unhealthy. Reason: ${reason}`); + poolManager.markProviderUnhealthyImmediately(MODEL_PROVIDER.KIRO_API, { + uuid: this.uuid + }, reason); + // Attach marker to error object to prevent duplicate marking in upper layers + if (error) { + error.credentialMarkedUnhealthy = true; + } + return true; + } else { + console.warn(`[Kiro] Cannot mark credential as unhealthy: poolManager=${!!poolManager}, uuid=${this.uuid}`); + return false; + } + } /** * Helper method to mark the current credential as unhealthy with a scheduled recovery time @@ -1822,7 +1847,7 @@ async saveCredentialsToFile(filePath, newData) { this.uuid = newUuid; } // 标记当前凭证为不健康（会自动进入刷新队列） - this._markCredentialUnhealthy('401 Unauthorized in stream - Triggering auto-refresh'); + this._markCredentialNeedRefresh('401 Unauthorized in stream - Triggering auto-refresh'); // Mark error for credential switch without recording error count error.shouldSwitchCredential = true; error.skipErrorCount = true; @@ -1837,7 +1862,7 @@ async saveCredentialsToFile(filePath, newData) { // Handle 403 (Forbidden) - mark as unhealthy immediately, no retry if (status === 403) { console.log('[Kiro] Received 403 in stream. Marking credential as unhealthy...'); - // this._markCredentialUnhealthy('403 Forbidden', error); + this._markCredentialUnhealthy('403 Forbidden', error); // Mark error for credential switch without recording error count error.shouldSwitchCredential = true; error.skipErrorCount = true; @@ -2723,13 +2748,13 @@ async saveCredentialsToFile(filePath, newData) { // 对于用量查询，401/403 错误直接标记凭证为不健康，不重试 if (status === 401) { console.log('[Kiro] Received 401 on getUsageLimits. Marking credential as unhealthy (no retry)...'); - this._markCredentialUnhealthy('401 Unauthorized on usage query', formattedError); + this._markCredentialNeedRefresh('401 Unauthorized on usage query', formattedError); throw formattedError; } if (status === 403) { console.log('[Kiro] Received 403 on getUsageLimits. Marking credential as unhealthy (no retry)...'); - // this._markCredentialUnhealthy('403 Forbidden on usage query', formattedError); + this._markCredentialUnhealthy('403 Forbidden on usage query', formattedError); throw formattedError; } diff --git a/src/providers/gemini/antigravity-core.js b/src/providers/gemini/antigravity-core.js index c2e6412..6817904 100644 --- a/src/providers/gemini/antigravity-core.js +++ b/src/providers/gemini/antigravity-core.js @@ -14,6 +14,8 @@ import { getProviderModels } from '../provider-models.js'; import { handleGeminiAntigravityOAuth } from '../../auth/oauth-handlers.js'; import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js'; import { cleanJsonSchemaProperties } from '../../converters/utils.js'; +import { getProviderPoolManager } from '../../services/service-manager.js'; +import { MODEL_PROVIDER } from '../../utils/common.js'; // 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏 const httpAgent = new http.Agent({ @@ -761,7 +763,9 @@ export class AntigravityApiService { async initialize() { if (this.isInitialized) return; console.log('[Antigravity] Initializing Antigravity API Service...'); - await this.initializeAuth(); + // 注意：V2 读写分离架构下，初始化不再执行同步认证/刷新逻辑 + // 仅执行基础的凭证加载 + await this.loadCredentials(); if (!this.projectId) { this.projectId = await this.discoverProjectAndModels(); @@ -775,6 +779,25 @@ export class AntigravityApiService { console.log(`[Antigravity] Initialization complete. Project ID: ${this.projectId}`); } + /** + * 加载凭证信息（不执行刷新） + */ + async loadCredentials() { + const credPath = this.oauthCredsFilePath || path.join(os.homedir(), CREDENTIALS_DIR, CREDENTIALS_FILE); + try { + const data = await fs.readFile(credPath, "utf8"); + const credentials = JSON.parse(data); + this.authClient.setCredentials(credentials); + console.log('[Antigravity Auth] Credentials loaded successfully from file.'); + } catch (error) { + if (error.code === 'ENOENT') { + console.debug(`[Antigravity Auth] Credentials file not found: ${credPath}`); + } else { + console.warn(`[Antigravity Auth] Failed to load credentials from file: ${error.message}`); + } + } + } + async initializeAuth(forceRefresh = false) { // 检查是否需要刷新 Token const needsRefresh = forceRefresh || this.isTokenExpiringSoon(); @@ -785,28 +808,40 @@ export class AntigravityApiService { } const credPath = this.oauthCredsFilePath || path.join(os.homedir(), CREDENTIALS_DIR, CREDENTIALS_FILE); - try { - const data = await fs.readFile(credPath, "utf8"); - const credentials = JSON.parse(data); - this.authClient.setCredentials(credentials); - console.log('[Antigravity Auth] Authentication configured successfully from file.'); + + // 首先执行基础凭证加载 + await this.loadCredentials(); - if (needsRefresh) { - console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...'); - const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); - this.authClient.setCredentials(newCredentials); - await this._saveCredentialsToFile(credPath, newCredentials); - console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`); - } - } catch (error) { - console.error('[Antigravity Auth] Error initializing authentication:', error.code); - if (error.code === 'ENOENT' || error.code === 400) { - console.log(`[Antigravity Auth] Credentials file '${credPath}' not found. Starting new authentication flow...`); - const newTokens = await this.getNewToken(credPath); - this.authClient.setCredentials(newTokens); - console.log('[Antigravity Auth] New token obtained and loaded into memory.'); - } else { - console.error('[Antigravity Auth] Failed to initialize authentication from file:', error); + // 只有在明确要求刷新，或者 AccessToken 确实缺失时，才执行刷新/认证 + // 注意：在 V2 架构下，此方法主要由 PoolManager 的后台队列调用 + if (needsRefresh || !this.authClient.credentials.access_token) { + try { + if (this.authClient.credentials.refresh_token) { + console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...'); + const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); + this.authClient.setCredentials(newCredentials); + await this._saveCredentialsToFile(credPath, newCredentials); + console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`); + + // 刷新成功，重置 PoolManager 中的刷新状态并标记为健康 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.ANTIGRAVITY, this.uuid); + } + } else { + console.log(`[Antigravity Auth] No access token or refresh token. Starting new authentication flow...`); + const newTokens = await this.getNewToken(credPath); + this.authClient.setCredentials(newTokens); + console.log('[Antigravity Auth] New token obtained and loaded into memory.'); + + // 认证成功，重置状态 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.ANTIGRAVITY, this.uuid); + } + } + } catch (error) { + console.error('[Antigravity Auth] Failed to initialize authentication:', error); throw new Error(`Failed to load OAuth credentials.`); } } @@ -1072,9 +1107,22 @@ export class AntigravityApiService { console.error(`[Antigravity API] Error calling ${method} on ${baseURL}:`, status, error.message); if ((status === 400 || status === 401) && !isRetry) { - console.log('[Antigravity API] Received 401/400. Refreshing auth and retrying...'); - await this.initializeAuth(true); - return this.callApi(method, body, true, retryCount, baseURLIndex); + console.log('[Antigravity API] Received 401/400. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Antigravity] Marking credential ${this.uuid} as needs refresh. Reason: 401/400 Unauthorized`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.ANTIGRAVITY, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } if (status === 429) { @@ -1161,10 +1209,22 @@ export class AntigravityApiService { console.error(`[Antigravity API] Error during stream ${method} on ${baseURL}:`, status, error.message); if ((status === 400 || status === 401) && !isRetry) { - console.log('[Antigravity API] Received 401/400 during stream. Refreshing auth and retrying...'); - await this.initializeAuth(true); - yield* this.streamApi(method, body, true, retryCount, baseURLIndex); - return; + console.log('[Antigravity API] Received 401/400 during stream. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Antigravity] Marking credential ${this.uuid} as needs refresh. Reason: 401/400 Unauthorized in stream`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.ANTIGRAVITY, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } if (status === 429) { @@ -1341,11 +1401,12 @@ export class AntigravityApiService { async getUsageLimits() { if (!this.isInitialized) await this.initialize(); - // 检查 token 是否即将过期，如果是则先刷新 - if (this.isExpiryDateNear()) { - console.log('[Antigravity] Token is near expiry, refreshing before getUsageLimits request...'); - await this.initializeAuth(true); - } + // 注意：V2 架构下不再在 getUsageLimits 中同步刷新 token + // 如果 token 过期，PoolManager 后台会自动处理 + // if (this.isExpiryDateNear()) { + // console.log('[Antigravity] Token is near expiry, refreshing before getUsageLimits request...'); + // await this.initializeAuth(true); + // } try { const modelsWithQuotas = await this.getModelsWithQuotas(); diff --git a/src/providers/gemini/gemini-core.js b/src/providers/gemini/gemini-core.js index bc4d913..98aacc7 100644 --- a/src/providers/gemini/gemini-core.js +++ b/src/providers/gemini/gemini-core.js @@ -10,6 +10,8 @@ import { API_ACTIONS, formatExpiryTime, isRetryableNetworkError } from '../../ut import { getProviderModels } from '../provider-models.js'; import { handleGeminiCliOAuth } from '../../auth/oauth-handlers.js'; import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js'; +import { getProviderPoolManager } from '../../services/service-manager.js'; +import { MODEL_PROVIDER } from '../../utils/common.js'; // 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏 const httpAgent = new http.Agent({ @@ -232,7 +234,10 @@ export class GeminiApiService { async initialize() { if (this.isInitialized) return; console.log('[Gemini] Initializing Gemini API Service...'); - await this.initializeAuth(); + // 注意：V2 读写分离架构下，初始化不再执行同步认证/刷新逻辑 + // 仅执行基础的凭证加载 + await this.loadCredentials(); + if (!this.projectId) { this.projectId = await this.discoverProjectAndModels(); } else { @@ -247,19 +252,19 @@ export class GeminiApiService { console.log(`[Gemini] Initialization complete. Project ID: ${this.projectId}`); } - async initializeAuth(forceRefresh = false) { - if (this.authClient.credentials.access_token && !forceRefresh) return; - + /** + * 加载凭证信息（不执行刷新） + */ + async loadCredentials() { if (this.oauthCredsBase64) { try { const decoded = Buffer.from(this.oauthCredsBase64, 'base64').toString('utf8'); const credentials = JSON.parse(decoded); this.authClient.setCredentials(credentials); - console.log('[Gemini Auth] Authentication configured successfully from base64 string.'); + console.log('[Gemini Auth] Credentials loaded successfully from base64 string.'); return; } catch (error) { console.error('[Gemini Auth] Failed to parse base64 OAuth credentials:', error); - throw new Error(`Failed to load OAuth credentials from base64 string.`); } } @@ -268,25 +273,65 @@ export class GeminiApiService { const data = await fs.readFile(credPath, "utf8"); const credentials = JSON.parse(data); this.authClient.setCredentials(credentials); - console.log('[Gemini Auth] Authentication configured successfully from file.'); - - if (forceRefresh) { - console.log('[Gemini Auth] Forcing token refresh...'); - const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); - this.authClient.setCredentials(newCredentials); - // Save refreshed credentials back to file - await this._saveCredentialsToFile(credPath, newCredentials); - console.log('[Gemini Auth] Token refreshed and saved successfully.'); - } + console.log('[Gemini Auth] Credentials loaded successfully from file.'); } catch (error) { - console.error('[Gemini Auth] Error initializing authentication:', error.code); - if (error.code === 'ENOENT' || error.code === 400) { - console.log(`[Gemini Auth] Credentials file '${credPath}' not found. Starting new authentication flow...`); - const newTokens = await this.getNewToken(credPath); - this.authClient.setCredentials(newTokens); - console.log('[Gemini Auth] New token obtained and loaded into memory.'); + if (error.code === 'ENOENT') { + console.debug(`[Gemini Auth] Credentials file not found: ${credPath}`); } else { - console.error('[Gemini Auth] Failed to initialize authentication from file:', error); + console.warn(`[Gemini Auth] Failed to load credentials from file: ${error.message}`); + } + } + } + + async initializeAuth(forceRefresh = false) { + // 检查是否需要刷新 Token + const needsRefresh = forceRefresh || this.isExpiryDateNear(); + + if (this.authClient.credentials.access_token && !needsRefresh) { + // Token 有效且不需要刷新 + return; + } + + // 首先执行基础凭证加载 + await this.loadCredentials(); + + // 只有在明确要求刷新，或者 AccessToken 确实缺失时，才执行刷新/认证 + // 注意：在 V2 架构下，此方法主要由 PoolManager 的后台队列调用 + if (needsRefresh || !this.authClient.credentials.access_token) { + const credPath = this.oauthCredsFilePath || path.join(os.homedir(), CREDENTIALS_DIR, CREDENTIALS_FILE); + try { + if (this.authClient.credentials.refresh_token) { + console.log('[Gemini Auth] Token expiring soon or force refresh requested. Refreshing token...'); + const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); + this.authClient.setCredentials(newCredentials); + + // 如果不是从 base64 加载的，则保存到文件 + if (!this.oauthCredsBase64) { + await this._saveCredentialsToFile(credPath, newCredentials); + console.log('[Gemini Auth] Token refreshed and saved successfully.'); + } else { + console.log('[Gemini Auth] Token refreshed successfully (Base64 source).'); + } + + // 刷新成功，重置 PoolManager 中的刷新状态并标记为健康 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.GEMINI_CLI, this.uuid); + } + } else { + console.log(`[Gemini Auth] No access token or refresh token. Starting new authentication flow...`); + const newTokens = await this.getNewToken(credPath); + this.authClient.setCredentials(newTokens); + console.log('[Gemini Auth] New token obtained and loaded into memory.'); + + // 认证成功，重置状态 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.GEMINI_CLI, this.uuid); + } + } + } catch (error) { + console.error('[Gemini Auth] Failed to initialize authentication:', error); throw new Error(`Failed to load OAuth credentials.`); } } @@ -447,9 +492,22 @@ export class GeminiApiService { // Handle 401 (Unauthorized) - refresh auth and retry once if ((status === 400 || status === 401) && !isRetry) { - console.log('[Gemini API] Received 401/400. Refreshing auth and retrying...'); - await this.initializeAuth(true); - return this.callApi(method, body, true, retryCount); + console.log('[Gemini API] Received 401/400. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Gemini] Marking credential ${this.uuid} as needs refresh. Reason: 401/400 Unauthorized`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.GEMINI_CLI, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } // Handle 429 (Too Many Requests) with exponential backoff @@ -513,10 +571,22 @@ export class GeminiApiService { // Handle 401 (Unauthorized) - refresh auth and retry once if ((status === 400 || status === 401) && !isRetry) { - console.log('[Gemini API] Received 401/400 during stream. Refreshing auth and retrying...'); - await this.initializeAuth(true); - yield* this.streamApi(method, body, true, retryCount); - return; + console.log('[Gemini API] Received 401/400 during stream. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Gemini] Marking credential ${this.uuid} as needs refresh. Reason: 401/400 Unauthorized in stream`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.GEMINI_CLI, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } // Handle 429 (Too Many Requests) with exponential backoff @@ -643,11 +713,12 @@ export class GeminiApiService { async getUsageLimits() { if (!this.isInitialized) await this.initialize(); - // 检查 token 是否即将过期，如果是则先刷新 - if (this.isExpiryDateNear()) { - console.log('[Gemini] Token is near expiry, refreshing before getUsageLimits request...'); - await this.initializeAuth(true); - } + // 注意：V2 架构下不再在 getUsageLimits 中同步刷新 token + // 如果 token 过期，PoolManager 后台会自动处理 + // if (this.isExpiryDateNear()) { + // console.log('[Gemini] Token is near expiry, refreshing before getUsageLimits request...'); + // await this.initializeAuth(true); + // } try { const modelsWithQuotas = await this.getModelsWithQuotas(); diff --git a/src/providers/openai/codex-core.js b/src/providers/openai/codex-core.js index aae6df8..bb0e033 100644 --- a/src/providers/openai/codex-core.js +++ b/src/providers/openai/codex-core.js @@ -4,10 +4,11 @@ import { promises as fs } from 'fs'; import path from 'path'; import os from 'os'; import { refreshCodexTokensWithRetry } from '../../auth/oauth-handlers.js'; +import { getProviderPoolManager } from '../../services/service-manager.js'; +import { MODEL_PROVIDER } from '../../utils/common.js'; /** * Codex API 服务类 - * 处理与 Codex API 的通信 */ export class CodexApiService { constructor(config) { @@ -18,6 +19,7 @@ export class CodexApiService { this.accountId = null; this.email = null; this.expiresAt = null; + this.uuid = config.uuid; // 保存 uuid 用于号池管理 this.isInitialized = false; // 会话缓存管理 @@ -29,6 +31,20 @@ export class CodexApiService { * 初始化服务（加载凭据） */ async initialize() { + if (this.isInitialized) return; + console.log('[Codex] Initializing Codex API Service...'); + // 注意：V2 读写分离架构下，初始化不再执行同步认证/刷新逻辑 + // 仅执行基础的凭证加载 + await this.loadCredentials(); + + this.isInitialized = true; + console.log(`[Codex] Initialization complete. Account: ${this.email || 'unknown'}`); + } + + /** + * 加载凭证信息（不执行刷新） + */ + async loadCredentials() { const email = this.config.CODEX_EMAIL || 'default'; try { @@ -60,10 +76,10 @@ export class CodexApiService { creds = JSON.parse(await fs.readFile(credsPath, 'utf8')); } - this.accessToken = creds.access_token; - this.refreshToken = creds.refresh_token; - this.accountId = creds.account_id; - this.email = creds.email; + this.accessToken = creds.access_token; + this.refreshToken = creds.refresh_token; + this.accountId = creds.account_id; + this.email = creds.email; this.expiresAt = new Date(creds.expired); // 注意：字段名是 expired // 检查 token 是否需要刷新 @@ -75,8 +91,37 @@ export class CodexApiService { this.isInitialized = true; console.log(`[Codex] Initialized with account: ${this.email}`); } catch (error) { - console.error('[Codex] Initialization failed:', error.message); - throw error; + console.warn(`[Codex Auth] Failed to load credentials: ${error.message}`); + } + } + + /** + * 初始化认证并执行必要刷新 + */ + async initializeAuth(forceRefresh = false) { + // 首先执行基础凭证加载 + await this.loadCredentials(); + + // 检查 token 是否需要刷新 + const needsRefresh = forceRefresh; + + if (this.accessToken && !needsRefresh) { + return; + } + + // 只有在明确要求刷新，或者 AccessToken 缺失时，才执行刷新 + if (needsRefresh || !this.accessToken) { + if (!this.refreshToken) { + throw new Error('Codex credentials not found. Please authenticate first using OAuth.'); + } + console.log('[Codex] Token expiring soon or refresh requested, refreshing...'); + await this.refreshAccessToken(); + + // 刷新成功，重置 PoolManager 中的刷新状态并标记为健康 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.CODEX_API, this.uuid); + } } } @@ -101,17 +146,22 @@ export class CodexApiService { return this.parseNonStreamResponse(response.data); } catch (error) { if (error.response?.status === 401) { - // Token 过期，尝试刷新 - console.log('[Codex] 401 error, refreshing token...'); - await this.refreshAccessToken(); - // 重试请求 - const retryBody = this.prepareRequestBody(model, requestBody, false); - const retryHeaders = this.buildHeaders(retryBody.prompt_cache_key); - const retryResponse = await axios.post(url, retryBody, { - headers: retryHeaders, - timeout: 120000 - }); - return this.parseNonStreamResponse(retryResponse.data); + console.log('[Codex] Received 401. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Codex] Marking credential ${this.uuid} as needs refresh. Reason: 401 Unauthorized`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.CODEX_API, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } throw error; } @@ -129,11 +179,6 @@ export class CodexApiService { const body = this.prepareRequestBody(model, requestBody, true); const headers = this.buildHeaders(body.prompt_cache_key); - // 调试日志 - console.log('[Codex Debug] Request URL:', url); - console.log('[Codex Debug] Request Body:', JSON.stringify(body, null, 2)); - console.log('[Codex Debug] Request Headers:', JSON.stringify(headers, null, 2)); - try { const response = await axios.post(url, body, { headers, @@ -143,33 +188,23 @@ export class CodexApiService { yield* this.parseSSEStream(response.data); } catch (error) { - // 打印详细错误信息 - if (error.response) { - console.error('[Codex Error] Status:', error.response.status); - console.error('[Codex Error] Headers:', error.response.headers); - if (error.response.data) { - const errorData = await new Promise((resolve) => { - let data = ''; - error.response.data.on('data', chunk => data += chunk); - error.response.data.on('end', () => resolve(data)); - }); - console.error('[Codex Error] Response:', errorData); - } - } - if (error.response?.status === 401) { - // Token 过期，尝试刷新 - console.log('[Codex] 401 error, refreshing token...'); - await this.refreshAccessToken(); - // 重试请求 - const retryBody = this.prepareRequestBody(model, requestBody, true); - const retryHeaders = this.buildHeaders(retryBody.prompt_cache_key); - const retryResponse = await axios.post(url, retryBody, { - headers: retryHeaders, - responseType: 'stream', - timeout: 120000 - }); - yield* this.parseSSEStream(retryResponse.data); + console.log('[Codex] Received 401 during stream. Triggering background refresh via PoolManager...'); + + // 标记当前凭证为不健康 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Codex] Marking credential ${this.uuid} as needs refresh. Reason: 401 Unauthorized in stream`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.CODEX_API, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; + } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } else { throw error; } diff --git a/src/providers/openai/iflow-core.js b/src/providers/openai/iflow-core.js index 6272d4e..8e13086 100644 --- a/src/providers/openai/iflow-core.js +++ b/src/providers/openai/iflow-core.js @@ -23,7 +23,8 @@ import { promises as fs } from 'fs'; import * as path from 'path'; import * as os from 'os'; import { configureAxiosProxy } from '../../utils/proxy-utils.js'; -import { isRetryableNetworkError } from '../../utils/common.js'; +import { isRetryableNetworkError, MODEL_PROVIDER } from '../../utils/common.js'; +import { getProviderPoolManager } from '../../services/service-manager.js'; // iFlow API 端点 const IFLOW_API_BASE_URL = 'https://apis.iflow.cn/v1'; @@ -493,18 +494,41 @@ export class IFlowApiService { if (this.isInitialized) return; console.log('[iFlow] Initializing iFlow API Service...'); - await this.initializeAuth(); + // 注意：V2 读写分离架构下，初始化不再执行同步认证/刷新逻辑 + // 仅执行基础的凭证加载 + await this.loadCredentials(); this.isInitialized = true; console.log('[iFlow] Initialization complete.'); } /** - * 初始化认证 + * 加载凭证信息（不执行刷新） + */ + async loadCredentials() { + try { + // 从文件加载 + this.tokenStorage = await loadTokenFromFile(this.tokenFilePath); + if (this.tokenStorage && this.tokenStorage.apiKey) { + this.apiKey = this.tokenStorage.apiKey; + // 更新 axios 实例的 Authorization header + this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`; + console.log('[iFlow Auth] Credentials loaded successfully from file'); + } + } catch (error) { + console.warn(`[iFlow Auth] Failed to load credentials from file: ${error.message}`); + } + } + + /** + * 初始化认证并执行必要刷新 * @param {boolean} forceRefresh - 是否强制刷新 Token */ async initializeAuth(forceRefresh = false) { - // 如果已有 API Key 且不强制刷新，直接返回 + // 首先执行基础凭证加载 + await this.loadCredentials(); + + // 如果已有 API Key 且不强制刷新且未过期，直接返回 if (this.apiKey && !forceRefresh) return; // 从 Token 文件加载 API Key @@ -527,23 +551,20 @@ export class IFlowApiService { console.log('[iFlow Auth] Forcing token refresh...'); await this._refreshOAuthTokens(); console.log('[iFlow Auth] Token refreshed and saved successfully.'); + + // 刷新成功，重置 PoolManager 中的刷新状态并标记为健康 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.IFLOW_API, this.uuid); + } } } else { - throw new Error('[iFlow] Token file does not contain a valid API key.'); + throw new Error('[iFlow] No refresh token available in credentials.'); } } catch (error) { - console.error('[iFlow Auth] Error initializing authentication:', error.code || error.message); - if (error.code === 'ENOENT') { - console.log(`[iFlow Auth] Credentials file '${this.tokenFilePath}' not found.`); - throw new Error(`[iFlow Auth] Credentials file not found. Please run OAuth flow first.`); - } else { - console.error('[iFlow Auth] Failed to initialize authentication from file:', error.message); - throw new Error(`[iFlow Auth] Failed to load OAuth credentials.`); - } + console.error('[iFlow Auth] Failed to initialize authentication:', error.message); + throw new Error(`[iFlow Auth] Failed to load OAuth credentials.`); } - - // 更新 axios 实例的 Authorization header - this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`; } /** @@ -562,10 +583,10 @@ export class IFlowApiService { } // 使用 isExpiryDateNear 检查过期时间 - if (!this.isExpiryDateNear()) { - console.log('[iFlow] Token is valid, no refresh needed'); - return false; - } + // if (!this.isExpiryDateNear()) { + // console.log('[iFlow] Token is valid, no refresh needed'); + // return false; + // } console.log('[iFlow] Token is expiring soon, attempting refresh...'); @@ -769,14 +790,22 @@ export class IFlowApiService { // Handle 401/400 - refresh auth and retry once if ((status === 400 || status === 401) && !isRetry) { - console.log(`[iFlow] Received ${status}. Refreshing auth and retrying...`); - try { - await this.initializeAuth(true); - return this.callApi(endpoint, body, model, true, retryCount); - } catch (authError) { - console.error('[iFlow] Failed to refresh auth during retry:', authError.message); - throw error; // throw original error if refresh fails + console.log(`[iFlow] Received ${status}. Triggering background refresh via PoolManager...`); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[iFlow] Marking credential ${this.uuid} as needs refresh. Reason: ${status} Unauthorized`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.IFLOW_API, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } if (status === 401 || status === 403) { @@ -919,15 +948,22 @@ export class IFlowApiService { // Handle 401/400 during stream - refresh auth and retry once if ((status === 400 || status === 401) && !isRetry) { - console.log(`[iFlow] Received ${status} during stream. Refreshing auth and retrying...`); - try { - await this.initializeAuth(true); - yield* this.streamApi(endpoint, body, model, true, retryCount); - return; - } catch (authError) { - console.error('[iFlow] Failed to refresh auth during stream retry:', authError.message); - throw error; + console.log(`[iFlow] Received ${status} during stream. Triggering background refresh via PoolManager...`); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[iFlow] Marking credential ${this.uuid} as needs refresh. Reason: ${status} Unauthorized in stream`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.IFLOW_API, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } if (status === 401 || status === 403) { @@ -976,8 +1012,8 @@ export class IFlowApiService { await this.initialize(); } - // 在 API 调用前检查是否需要刷新 Token - await this._checkAndRefreshTokenIfNeeded(); + // 在 API 调用前不再同步检查是否需要刷新 Token (V2 架构) + // await this._checkAndRefreshTokenIfNeeded(); return this.callApi('/chat/completions', requestBody, model); } @@ -990,8 +1026,8 @@ export class IFlowApiService { await this.initialize(); } - // 在 API 调用前检查是否需要刷新 Token - await this._checkAndRefreshTokenIfNeeded(); + // 在 API 调用前不再同步检查是否需要刷新 Token (V2 架构) + // await this._checkAndRefreshTokenIfNeeded(); yield* this.streamApi('/chat/completions', requestBody, model); } diff --git a/src/providers/openai/qwen-core.js b/src/providers/openai/qwen-core.js index cab299f..4d5b454 100644 --- a/src/providers/openai/qwen-core.js +++ b/src/providers/openai/qwen-core.js @@ -11,7 +11,8 @@ import { randomUUID } from 'node:crypto'; import { getProviderModels } from '../provider-models.js'; import { handleQwenOAuth } from '../../auth/oauth-handlers.js'; import { configureAxiosProxy } from '../../utils/proxy-utils.js'; -import { isRetryableNetworkError } from '../../utils/common.js'; +import { isRetryableNetworkError, MODEL_PROVIDER } from '../../utils/common.js'; +import { getProviderPoolManager } from '../../services/service-manager.js'; // --- Constants --- const QWEN_DIR = '.qwen'; @@ -179,6 +180,7 @@ export class QwenApiService { this.currentAxiosInstance = null; this.tokenManagerOptions = { credentialFilePath: this._getQwenCachedCredentialPath() }; this.useSystemProxy = config?.USE_SYSTEM_PROXY_QWEN ?? false; + this.uuid = config.uuid; // 保存 uuid 用于号池管理 // Initialize instance-specific endpoints this.baseUrl = config.QWEN_BASE_URL || DEFAULT_QWEN_BASE_URL; @@ -193,7 +195,9 @@ export class QwenApiService { async initialize() { if (this.isInitialized) return; console.log('[Qwen] Initializing Qwen API Service...'); - await this._initializeAuth(); + // 注意：V2 读写分离架构下，初始化不再执行同步认证/刷新逻辑 + // 仅执行基础的凭证加载 + await this.loadCredentials(); // 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏 const httpAgent = new http.Agent({ @@ -233,7 +237,29 @@ export class QwenApiService { console.log('[Qwen] Initialization complete.'); } + /** + * 加载凭证信息（不执行刷新） + */ + async loadCredentials() { + try { + const keyFile = this._getQwenCachedCredentialPath(); + const creds = await fs.readFile(keyFile, 'utf-8'); + const credentials = JSON.parse(creds); + this.qwenClient.setCredentials(credentials); + console.log('[Qwen Auth] Credentials loaded successfully from file.'); + } catch (error) { + if (error.code === 'ENOENT') { + console.debug('[Qwen Auth] No cached credentials found.'); + } else { + console.warn(`[Qwen Auth] Failed to load credentials from file: ${error.message}`); + } + } + } + async _initializeAuth(forceRefresh = false) { + // 首先执行基础凭证加载 + await this.loadCredentials(); + try { const credentials = await this.sharedManager.getValidCredentials( this.qwenClient, @@ -242,6 +268,14 @@ export class QwenApiService { ); // console.log('credentials', credentials); this.qwenClient.setCredentials(credentials); + + // 如果执行了刷新或认证，重置状态 + if (forceRefresh || (credentials && credentials.access_token)) { + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.QWEN_API, this.uuid); + } + } } catch (error) { console.debug('Shared token manager failed, attempting device flow:', error); @@ -288,9 +322,22 @@ export class QwenApiService { default: throw new Error('Qwen OAuth authentication failed'); } + } else { + // 认证成功，重置状态 + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + poolManager.resetProviderRefreshStatus(MODEL_PROVIDER.QWEN_API, this.uuid); + } } } } + + /** + * 实现与其它 provider 统一的 initializeAuth 接口 + */ + async initializeAuth(forceRefresh = false) { + return this._initializeAuth(forceRefresh); + } async _authWithQwenDeviceFlow(client, config) { try { @@ -563,18 +610,22 @@ export class QwenApiService { const isNetworkError = isRetryableNetworkError(error); if (this.isAuthError(error) && retryCount === 0) { - console.warn(`[QwenApiService] Auth error (${status}). Refreshing token...`); - try { - await this.sharedManager.getValidCredentials( - this.qwenClient, - true, - this.tokenManagerOptions, - ); - return this.callApiWithAuthAndRetry(endpoint, body, isStream, retryCount + 1); - } catch (refreshError) { - console.error(`[QwenApiService] Token refresh failed:`, refreshError); - throw new Error(`Token refresh failed. Please re-authenticate. ${refreshError.message}`); + console.warn(`[QwenApiService] Auth error (${status}). Triggering background refresh via PoolManager...`); + + // 标记当前凭证为不健康（会自动进入刷新队列） + const poolManager = getProviderPoolManager(); + if (poolManager && this.uuid) { + console.log(`[Qwen] Marking credential ${this.uuid} as needs refresh. Reason: Auth Error ${status}`); + poolManager.markProviderNeedRefresh(MODEL_PROVIDER.QWEN_API, { + uuid: this.uuid + }); + error.credentialMarkedUnhealthy = true; } + + // Mark error for credential switch without recording error count + error.shouldSwitchCredential = true; + error.skipErrorCount = true; + throw error; } if ((status === 429 || (status >= 500 && status < 600)) && retryCount < maxRetries) { diff --git a/src/providers/provider-models.js b/src/providers/provider-models.js index 8d841cc..2b6ac3d 100644 --- a/src/providers/provider-models.js +++ b/src/providers/provider-models.js @@ -25,9 +25,9 @@ export const PROVIDER_MODELS = { ], 'claude-custom': [], 'claude-kiro-oauth': [ + 'claude-haiku-4-5', 'claude-opus-4-5', 'claude-opus-4-5-20251101', - 'claude-haiku-4-5', 'claude-sonnet-4-5', 'claude-sonnet-4-5-20250929', 'claude-sonnet-4-20250514',