From 2706c07381c3177c4e8fc395a78dcbd8b68b68f2 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 12 Jul 2026 14:17:27 +0800 Subject: [PATCH] feat(desktop-monitor): multi-account pool with failover and cooldown Add monitor-account-pool to select authorized, non-blocked candidates for an AI-platform monitor task, classify failover reasons (risk control, challenge, auth failure, empty/unknown result) and cool the account down before rotating to the next. Wire runtime-controller to run monitor tasks through the pool, retrying across accounts and annotating the result with per-account attempt diagnostics and a failover count. Expose the pool state via the runtime snapshot. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/main/monitor-account-pool.test.ts | 112 ++++++++ .../src/main/monitor-account-pool.ts | 162 ++++++++++++ .../src/main/runtime-controller.ts | 248 ++++++++++++++++-- .../src/main/runtime-snapshot.ts | 2 + 4 files changed, 501 insertions(+), 23 deletions(-) create mode 100644 apps/desktop-client/src/main/monitor-account-pool.test.ts create mode 100644 apps/desktop-client/src/main/monitor-account-pool.ts diff --git a/apps/desktop-client/src/main/monitor-account-pool.test.ts b/apps/desktop-client/src/main/monitor-account-pool.test.ts new file mode 100644 index 0000000..2d241c1 --- /dev/null +++ b/apps/desktop-client/src/main/monitor-account-pool.test.ts @@ -0,0 +1,112 @@ +import { beforeEach, describe, expect, it } from 'vitest' + +import { + classifyMonitorAccountFailover, + coolDownMonitorAccount, + noteMonitorAccountAttempt, + resetMonitorAccountPoolState, + selectMonitorAccountCandidates, +} from './monitor-account-pool' + +describe('monitor account pool', () => { + beforeEach(() => resetMonitorAccountPoolState()) + + it('uses the server-preferred account first and least-recently-used alternatives next', () => { + const entries = [ + { id: 'doubao-a', authState: 'active', hasLocalAuthorization: true }, + { id: 'doubao-b', authState: 'active', hasLocalAuthorization: true }, + { id: 'doubao-c', authState: 'active', hasLocalAuthorization: true }, + ] + noteMonitorAccountAttempt('doubao-b', 200) + noteMonitorAccountAttempt('doubao-c', 100) + + expect(selectMonitorAccountCandidates(entries, 'doubao-a', 300).map((item) => item.id)).toEqual( + ['doubao-a', 'doubao-c', 'doubao-b'], + ) + }) + + it('skips challenged, expired, missing-session, and cooling accounts', () => { + const entries = [ + { id: 'active', authState: 'active', hasLocalAuthorization: true }, + { id: 'risk', authState: 'challenge_required', hasLocalAuthorization: true }, + { id: 'expired', authState: 'expired', hasLocalAuthorization: true }, + { id: 'missing', authState: 'active', hasLocalAuthorization: false }, + { id: 'cooling', authState: 'active', hasLocalAuthorization: true }, + ] + coolDownMonitorAccount('cooling', 'empty_result', 1_000) + + expect(selectMonitorAccountCandidates(entries, null, 1_001).map((item) => item.id)).toEqual([ + 'active', + ]) + }) + + it('releases a temporarily cooling account after its cooldown', () => { + const entries = [{ id: 'account', authState: 'active', hasLocalAuthorization: true }] + coolDownMonitorAccount('account', 'adapter_failure', 10_000) + + expect(selectMonitorAccountCandidates(entries, null, 10_001)).toEqual([]) + expect(selectMonitorAccountCandidates(entries, null, 70_001)).toHaveLength(1) + }) + + it('fails over for risk control, unknown, empty, and source-less successful responses', () => { + expect( + classifyMonitorAccountFailover({ + status: 'failed', + summary: '豆包触发风控', + error: { code: 'doubao_challenge_required', message: 'risk_control' }, + }), + ).toBe('risk_control') + expect(classifyMonitorAccountFailover({ status: 'unknown', summary: '未拿到结果' })).toBe( + 'unknown_result', + ) + expect( + classifyMonitorAccountFailover({ status: 'succeeded', summary: '完成', payload: {} }), + ).toBe('empty_result') + expect( + classifyMonitorAccountFailover({ + status: 'succeeded', + summary: '完成', + payload: { answer: '有效回答' }, + }), + ).toBe('missing_citations') + expect( + classifyMonitorAccountFailover({ + status: 'succeeded', + summary: '完成', + payload: { + answer: '有引用的回答', + citations: [{ url: 'https://example.com/citation', title: '引用来源' }], + search_results: [], + }, + }), + ).toBeNull() + expect( + classifyMonitorAccountFailover({ + status: 'succeeded', + summary: '完成', + payload: { + answer: '有搜索来源的回答', + citations: [], + search_results: [{ url: 'https://example.com/search', title: '搜索来源' }], + }, + }), + ).toBeNull() + }) + + it('does not rotate accounts for deterministic task or infrastructure failures', () => { + expect( + classifyMonitorAccountFailover({ + status: 'failed', + summary: 'question missing', + error: { code: 'doubao_question_text_missing' }, + }), + ).toBeNull() + expect( + classifyMonitorAccountFailover({ + status: 'failed', + summary: 'Playwright unavailable', + error: { code: 'playwright_cdp_unavailable' }, + }), + ).toBeNull() + }) +}) diff --git a/apps/desktop-client/src/main/monitor-account-pool.ts b/apps/desktop-client/src/main/monitor-account-pool.ts new file mode 100644 index 0000000..e7bbd64 --- /dev/null +++ b/apps/desktop-client/src/main/monitor-account-pool.ts @@ -0,0 +1,162 @@ +import type { AdapterExecutionResult } from './adapters/base' + +export type MonitorAccountFailoverReason = + | 'risk_control' + | 'challenge_required' + | 'authorization_failed' + | 'empty_result' + | 'missing_citations' + | 'unknown_result' + | 'adapter_failure' + +export type MonitorAccountPoolEntry = { + id: string + authState: string + hasLocalAuthorization: boolean +} + +type MonitorAccountCooldown = { + reason: MonitorAccountFailoverReason + until: number +} + +const lastUsedAt = new Map() +const cooldowns = new Map() + +const blockedAuthStates = new Set(['challenge_required', 'expired', 'revoked']) +const deterministicFailurePattern = + /adapter_aborted|stale_business_date|desktop_monitor_task_stale|desktop_task_adapter_missing|question_text_missing|playwright_required|playwright_cdp_(?:unavailable|disabled)/i +const riskControlPattern = /risk[_ -]?control|风控|频繁|访问受限|操作受限/i +const challengePattern = /challenge|required|captcha|验证码|人机验证|人工验证|安全验证/i +const authorizationPattern = + /authorization|unauthorized|login[_ -]?(?:required|expired)|not logged in|登录态|未登录|请先登录|授权过期/i + +export function resetMonitorAccountPoolState(): void { + lastUsedAt.clear() + cooldowns.clear() +} + +export function isMonitorAccountAuthBlocked(authState: string): boolean { + return blockedAuthStates.has(authState) +} + +export function selectMonitorAccountCandidates( + entries: readonly MonitorAccountPoolEntry[], + preferredAccountId: string | null, + now = Date.now(), +): MonitorAccountPoolEntry[] { + for (const [accountId, cooldown] of cooldowns.entries()) { + if (cooldown.until <= now) { + cooldowns.delete(accountId) + } + } + + return entries + .filter( + (entry) => + entry.hasLocalAuthorization && + !isMonitorAccountAuthBlocked(entry.authState) && + (cooldowns.get(entry.id)?.until ?? 0) <= now, + ) + .sort((left, right) => { + if (left.id === preferredAccountId && right.id !== preferredAccountId) { + return -1 + } + if (right.id === preferredAccountId && left.id !== preferredAccountId) { + return 1 + } + const usedDelta = (lastUsedAt.get(left.id) ?? 0) - (lastUsedAt.get(right.id) ?? 0) + return usedDelta || left.id.localeCompare(right.id) + }) +} + +export function noteMonitorAccountAttempt(accountId: string, now = Date.now()): void { + lastUsedAt.set(accountId, now) +} + +export function clearMonitorAccountCooldown(accountId: string): void { + cooldowns.delete(accountId) +} + +export function monitorAccountCooldownMs(reason: MonitorAccountFailoverReason): number { + switch (reason) { + case 'risk_control': + case 'challenge_required': + case 'authorization_failed': + return 30 * 60_000 + case 'empty_result': + case 'unknown_result': + return 3 * 60_000 + case 'missing_citations': + return 0 + case 'adapter_failure': + return 60_000 + } +} + +export function coolDownMonitorAccount( + accountId: string, + reason: MonitorAccountFailoverReason, + now = Date.now(), +): void { + cooldowns.set(accountId, { + reason, + until: now + monitorAccountCooldownMs(reason), + }) +} + +export function classifyMonitorAccountFailover( + result: AdapterExecutionResult, +): MonitorAccountFailoverReason | null { + const answerCandidates = [ + result.payload?.answer, + result.payload?.answer_text, + result.payload?.content, + result.payload?.response_text, + ] + const hasAnswer = answerCandidates.some( + (value) => typeof value === 'string' && value.trim().length > 0, + ) + if (result.status === 'succeeded' && hasAnswer) { + const hasCitationSources = [result.payload?.citations, result.payload?.search_results].some( + (value) => Array.isArray(value) && value.length > 0, + ) + return hasCitationSources ? null : 'missing_citations' + } + if (result.status === 'succeeded') { + return 'empty_result' + } + if (result.status === 'unknown') { + return 'unknown_result' + } + + const errorCode = typeof result.error?.code === 'string' ? result.error.code : '' + const errorMessage = typeof result.error?.message === 'string' ? result.error.message : '' + const failureText = `${errorCode} ${errorMessage} ${result.summary}` + if (deterministicFailurePattern.test(failureText)) { + return null + } + if (riskControlPattern.test(failureText)) { + return 'risk_control' + } + if (challengePattern.test(failureText)) { + return 'challenge_required' + } + if (authorizationPattern.test(failureText)) { + return 'authorization_failed' + } + return 'adapter_failure' +} + +export function getMonitorAccountPoolSnapshot(now = Date.now()) { + return { + trackedAccountCount: lastUsedAt.size, + cooldowns: [...cooldowns.entries()] + .filter(([, cooldown]) => cooldown.until > now) + .map(([accountId, cooldown]) => ({ + accountId, + reason: cooldown.reason, + until: cooldown.until, + })), + } +} diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index b48a60c..d12d992 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -61,6 +61,15 @@ import { noteLeaseReleased, setActiveLease, } from './lease-manager' +import { + classifyMonitorAccountFailover, + clearMonitorAccountCooldown, + coolDownMonitorAccount, + isMonitorAccountAuthBlocked, + noteMonitorAccountAttempt, + selectMonitorAccountCandidates, + type MonitorAccountFailoverReason, +} from './monitor-account-pool' import { enqueueMonitorLeaseTask, getMonitorSchedulerSnapshot, @@ -395,10 +404,6 @@ function shouldReportAccountPresence(account: DesktopAccountInfo): boolean { } function buildRuntimeAccountDedupeKey(account: DesktopAccountInfo): string { - if (isAIPlatformId(account.platform)) { - return `ai-platform:${account.platform}` - } - return buildAccountIdentityKey({ platform: account.platform, platformUid: account.platform_uid, @@ -466,26 +471,34 @@ function accountActionRequiredSummary( } function monitorPlatformBlockedReason(platform: string): string | null { - const account = state.accounts.find( + const ownedAccounts = state.accounts.filter( (item) => item.platform === platform && isAccountOwnedByCurrentClient(item), ) - if (!account) { + if (ownedAccounts.length === 0) { return null } + const accounts = ownedAccounts.filter((item) => hasLocalAccountAuthorization(item)) + if (accounts.length === 0) { + return `${runtimePlatformLabel(platform)} 本地账号会话不可用，请重新授权后再继续执行。` + } - const projected = getProjectedAccountHealth({ - accountId: account.id, - platform: account.platform, - health: account.health, - verifiedAt: account.verified_at, - }) - if (projected.authState === 'challenge_required') { - return accountActionRequiredSummary(account.platform, projected.authReason) + let firstBlockedReason: string | null = null + for (const account of accounts) { + const projected = getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }) + if (!isMonitorAccountAuthBlocked(projected.authState)) { + return null + } + firstBlockedReason ??= + projected.authState === 'challenge_required' + ? accountActionRequiredSummary(account.platform, projected.authReason) + : `${runtimePlatformLabel(account.platform)} 账号登录态已失效，请重新授权后再继续执行。` } - if (projected.authState === 'expired' || projected.authState === 'revoked') { - return `${runtimePlatformLabel(account.platform)} 账号登录态已失效，请重新授权后再继续执行。` - } - return null + return firstBlockedReason } function blockedMonitorPlatforms(): Set { @@ -574,6 +587,9 @@ function withAuthorizationRetryPolicy( async function resolveAuthorizationPreflightBlock( task: RuntimeTaskRecord, ): Promise { + if (task.kind === 'monitor' && isAIPlatformId(task.platform)) { + return null + } const accountIdentity = accountIdentityFromTask(task) if (!accountIdentity || !shouldEnforceActiveAccount(task)) { return null @@ -1669,11 +1685,14 @@ export async function requestRuntimeAccountProbe( } try { - await probeTrackedAccount(account, { + const snapshot = await probeTrackedAccount(account, { trigger: 'manual', allowSilentRefresh: true, force: options.force ?? true, }) + if (snapshot.authState === 'active') { + clearMonitorAccountCooldown(account.id) + } enqueueAccountHealthReport(account.id) await mirrorProjectedAccountProfile(account.id) } catch (error) { @@ -1697,8 +1716,6 @@ export function noteRuntimeAccountBound(account: DesktopAccountInfo): void { const existingIndex = state.accounts.findIndex( (item) => item.id === normalizedAccount.id || - (isAIPlatformId(normalizedAccount.platform) && - item.platform === normalizedAccount.platform) || (item.platform === normalizedAccount.platform && sameAccountPlatformUid(item.platform_uid, normalizedAccount.platform_uid)), ) @@ -1744,6 +1761,7 @@ export function noteRuntimeAccountBound(account: DesktopAccountInfo): void { displayName: normalizedAccount.display_name, avatarUrl: normalizedAccount.avatar_url, }) + clearMonitorAccountCooldown(normalizedAccount.id) enqueueAccountHealthReport(normalizedAccount.id) refreshAccountNames() } @@ -3132,7 +3150,8 @@ async function executeTaskAdapter( signal: AbortSignal, ): Promise { const accountIdentity = accountIdentityFromTask(task) - if (accountIdentity && shouldEnforceActiveAccount(task)) { + const usesMonitorAccountPool = task.kind === 'monitor' && isAIPlatformId(task.platform) + if (accountIdentity && shouldEnforceActiveAccount(task) && !usesMonitorAccountPool) { const readiness = await ensureAccountReady(accountIdentity) if (readiness.authState !== 'active') { return buildAccountBlockedResult(task, readiness) @@ -3217,6 +3236,190 @@ async function executeTaskAdapter( return buildScaffoldResult(task, payload, 'monitor adapter is not implemented yet') } + if (usesMonitorAccountPool) { + return await executeMonitorTaskWithAccountPool(task, adapter, payload, signal) + } + + const result = await executeMonitorAdapterAttempt(task, adapter, payload, signal) + await maybeReportTaskAuthFailure(task, result, accountIdentity) + return result +} + +type MonitorAccountAttemptDiagnostic = { + account_id: string + status: AdapterExecutionResult['status'] + failover_reason: MonitorAccountFailoverReason | null +} + +function withMonitorAccountDiagnostics( + result: AdapterExecutionResult, + attempts: readonly MonitorAccountAttemptDiagnostic[], + selectedAccountId: string | null, +): AdapterExecutionResult { + return { + ...result, + payload: { + ...(result.payload ?? {}), + execution_account_id: selectedAccountId, + account_attempt_count: attempts.length, + account_failover_count: Math.max(0, attempts.length - 1), + account_attempts: attempts.map((attempt) => ({ ...attempt })), + }, + } +} + +function monitorAccountPoolForTask(task: RuntimeTaskRecord): DesktopAccountInfo[] { + const platformAccounts = state.accounts.filter((account) => account.platform === task.platform) + const selected = selectMonitorAccountCandidates( + platformAccounts.map((account) => { + const projected = getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }) + return { + id: account.id, + authState: projected.authState, + hasLocalAuthorization: hasLocalAccountAuthorization(account), + } + }), + task.accountId || null, + ) + const byId = new Map(platformAccounts.map((account) => [account.id, account])) + return selected + .map((candidate) => byId.get(candidate.id) ?? null) + .filter((account): account is DesktopAccountInfo => account !== null) +} + +async function executeMonitorTaskWithAccountPool( + task: RuntimeTaskRecord, + adapter: MonitorAdapter, + payload: Record, + signal: AbortSignal, +): Promise { + const accounts = monitorAccountPoolForTask(task) + const attempts: MonitorAccountAttemptDiagnostic[] = [] + + if (accounts.length === 0) { + const blockedAccount = + state.accounts.find((account) => account.id === task.accountId) ?? + state.accounts.find((account) => account.platform === task.platform) ?? + null + if (blockedAccount) { + const identity = accountIdentityFromDesktopAccount(blockedAccount) + const previous = getAccountHealthSnapshot(identity.id) + const readiness = await ensureAccountReady(identity) + const blocked = buildAccountBlockedResult(task, readiness) + maybeRecordAccountAccessAlert(task, identity, previous, readiness) + enqueueAccountHealthReport(identity.id) + return withMonitorAccountDiagnostics(blocked, attempts, null) + } + + return withMonitorAccountDiagnostics( + { + status: 'failed', + summary: `${runtimePlatformLabel(task.platform)} 没有可用的本地账号，请先绑定或恢复账号后再采集。`, + error: markAuthorizationFailureNonRetryable( + { + code: 'desktop_monitor_account_pool_empty', + message: 'no usable local account is available for this monitor platform', + }, + 'ai_platform_authorization', + ), + }, + attempts, + null, + ) + } + + let lastResult: AdapterExecutionResult | null = null + let lastAccountId: string | null = null + + for (let index = 0; index < accounts.length; index += 1) { + const account = accounts[index] + const identity = accountIdentityFromDesktopAccount(account) + const attemptTask: RuntimeTaskRecord = { + ...task, + accountId: account.id, + accountName: account.display_name, + } + lastAccountId = account.id + noteMonitorAccountAttempt(account.id) + + if (index === 0 && account.id !== task.accountId) { + recordActivity( + 'warn', + '模型账号自动切换', + `${runtimePlatformLabel(task.platform)} 原定账号不可用，${task.title} 已切换到 ${account.display_name}。`, + ) + } + + const previousSnapshot = getAccountHealthSnapshot(identity.id) + const readiness = await ensureAccountReady(identity) + if (readiness.authState !== 'active') { + lastResult = buildAccountBlockedResult(attemptTask, readiness) + maybeRecordAccountAccessAlert(attemptTask, identity, previousSnapshot, readiness) + enqueueAccountHealthReport(identity.id) + } else { + try { + lastResult = await executeMonitorAdapterAttempt(attemptTask, adapter, payload, signal) + } catch (error) { + if (signal.aborted || isPlaywrightCDPInfrastructureError(error)) { + throw error + } + lastResult = { + status: 'failed', + summary: `${runtimePlatformLabel(task.platform)} 账号 ${account.display_name} 执行失败：${errorMessage(error)}`, + error: toStructuredError(error), + } + } + await maybeReportTaskAuthFailure(attemptTask, lastResult, identity) + } + + const failoverReason = classifyMonitorAccountFailover(lastResult) + attempts.push({ + account_id: account.id, + status: lastResult.status, + failover_reason: failoverReason, + }) + if (!failoverReason) { + clearMonitorAccountCooldown(account.id) + return withMonitorAccountDiagnostics(lastResult, attempts, account.id) + } + + coolDownMonitorAccount(account.id, failoverReason) + const nextAccount = accounts[index + 1] + if (nextAccount) { + recordActivity( + 'warn', + '模型账号自动切换', + `${runtimePlatformLabel(task.platform)} 账号 ${account.display_name} 本轮返回 ${failoverReason}，${task.title} 将改用 ${nextAccount.display_name} 继续采集。`, + ) + } + } + + return withMonitorAccountDiagnostics( + lastResult ?? { + status: 'failed', + summary: `${runtimePlatformLabel(task.platform)} 所有绑定账号均未返回有效结果。`, + error: { + code: 'desktop_monitor_account_pool_exhausted', + message: 'all same-platform monitor accounts were exhausted', + }, + }, + attempts, + lastAccountId, + ) +} + +async function executeMonitorAdapterAttempt( + task: RuntimeTaskRecord, + adapter: MonitorAdapter, + payload: Record, + signal: AbortSignal, +): Promise { + const sessionHandle = createSessionHandle(task.accountId || task.id) const viewHandle = adapter.executionMode === 'session' || adapter.executionMode === 'playwright' ? null @@ -3248,7 +3451,6 @@ async function executeTaskAdapter( }, payload, ) - await maybeReportTaskAuthFailure(task, result, accountIdentity) return result } finally { await playwrightLease?.release() diff --git a/apps/desktop-client/src/main/runtime-snapshot.ts b/apps/desktop-client/src/main/runtime-snapshot.ts index c289f68..2526ab1 100644 --- a/apps/desktop-client/src/main/runtime-snapshot.ts +++ b/apps/desktop-client/src/main/runtime-snapshot.ts @@ -6,6 +6,7 @@ import { getCircuitBreakerState } from './circuit-breaker' import { collectRecoverySnapshot } from './crash-recovery' import { getKeepAlivePlan } from './keep-alive' import { getLeaseManagerSnapshot } from './lease-manager' +import { getMonitorAccountPoolSnapshot } from './monitor-account-pool' import { getMonitorSchedulerSnapshot } from './monitor-scheduler' import { getObservedRequestSnapshot } from './network-observer' import { getHiddenPlaywrightSnapshot } from './playwright-cdp' @@ -206,6 +207,7 @@ function createLiveRuntimeSnapshot(controller: RuntimeControllerSnapshot) { circuitBreaker: getCircuitBreakerState(), keepAlive: getKeepAlivePlan(), leaseManager, + monitorAccountPool: getMonitorAccountPoolSnapshot(now), rateLimiter, recovery: collectRecoverySnapshot(), runtimeController: {