From aa9ce6f23e9ae451c348f18f10f7a92ee97af05a Mon Sep 17 00:00:00 2001 From: liangxu Date: Tue, 14 Jul 2026 22:01:48 +0800 Subject: [PATCH] feat(desktop): pause a platform after its last account hits a challenge Add a per-platform circuit breaker that trips when a monitor task's final account fails over with risk_control / challenge_required / authorization_failed. Tripping cancels the platform's remaining pending local monitor tasks (aborted with a cancellation summary) and blocks new ones until an account probes healthy or a fresh account is bound. Adds cancelPendingMonitorTasksForPlatform to the monitor scheduler plus tests. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/main/monitor-scheduler.test.ts | 44 ++++++++++++++ .../src/main/monitor-scheduler.ts | 26 +++++++++ .../src/main/runtime-controller.ts | 58 ++++++++++++++++++- 3 files changed, 127 insertions(+), 1 deletion(-) diff --git a/apps/desktop-client/src/main/monitor-scheduler.test.ts b/apps/desktop-client/src/main/monitor-scheduler.test.ts index 0289317..4800e9e 100644 --- a/apps/desktop-client/src/main/monitor-scheduler.test.ts +++ b/apps/desktop-client/src/main/monitor-scheduler.test.ts @@ -10,7 +10,9 @@ vi.mock('electron/main', () => ({ })) import { + cancelPendingMonitorTasksForPlatform, enqueueMonitorLeaseTask, + getMonitorSchedulerSnapshot, initMonitorScheduler, noteMonitorTaskActivated, noteMonitorTaskCompleted, @@ -132,4 +134,46 @@ describe('monitor scheduler', () => { })?.taskId, ).toBe('collect-now') }) + + it('removes later same-platform tasks after the only account hits a challenge', () => { + enqueueMonitorLeaseTask({ + taskId: 'doubao-current', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + questionKey: 'q:current', + }) + enqueueMonitorLeaseTask({ + taskId: 'doubao-next', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + questionKey: 'q:next', + }) + enqueueMonitorLeaseTask({ + taskId: 'qwen-next', + clientId: 'client-1', + platform: 'qwen', + routing: 'rabbitmq-primary', + questionKey: 'q:qwen', + }) + noteMonitorTaskActivated('doubao-current') + + expect(cancelPendingMonitorTasksForPlatform('doubao', 'doubao-current')).toEqual([ + expect.objectContaining({ taskId: 'doubao-next', platform: 'doubao' }), + ]) + expect( + getMonitorSchedulerSnapshot() + .tasks.map((task) => task.taskId) + .sort(), + ).toEqual(['doubao-current', 'qwen-next']) + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(['doubao']), + activeQuestionKeys: new Set(['q:current']), + activeCount: 1, + })?.taskId, + ).toBe('qwen-next') + }) }) diff --git a/apps/desktop-client/src/main/monitor-scheduler.ts b/apps/desktop-client/src/main/monitor-scheduler.ts index 8007a75..652f83d 100644 --- a/apps/desktop-client/src/main/monitor-scheduler.ts +++ b/apps/desktop-client/src/main/monitor-scheduler.ts @@ -474,6 +474,32 @@ export function noteMonitorTaskCanceled(taskId: string): void { }) } +export function cancelPendingMonitorTasksForPlatform( + platform: string, + currentTaskId?: string | null, +): MonitorSchedulerTaskRecord[] { + const normalizedPlatform = platform.trim() + if (!normalizedPlatform) { + return [] + } + + const canceled: MonitorSchedulerTaskRecord[] = [] + mutatePersistedState((draft) => { + for (const [taskId, task] of Object.entries(draft.tasks)) { + if ( + task.platform !== normalizedPlatform || + taskId === currentTaskId || + task.state === 'active' + ) { + continue + } + canceled.push({ ...task }) + delete draft.tasks[taskId] + } + }) + return canceled +} + export function selectNextMonitorTask( options: MonitorSchedulerSelectionOptions, ): MonitorSchedulerSelection | null { diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index b3b3c39..da097ba 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -71,6 +71,7 @@ import { type MonitorAccountFailoverReason, } from './monitor-account-pool' import { + cancelPendingMonitorTasksForPlatform, enqueueMonitorLeaseTask, getMonitorSchedulerSnapshot, initMonitorScheduler, @@ -191,6 +192,7 @@ const accountHealthReportDebounceMs = 1_000 const accountHealthReportMaxBatchSize = 100 const maxPlaywrightCDPSoftRecoveries = 3 const playwrightCDPFatalMarker = '--geo-cdp-fatal' +const monitorPlatformCircuitBreakers = new Set() interface RuntimeTaskRecord { id: string @@ -471,7 +473,7 @@ function accountActionRequiredSummary( return `${label} 账号需要完成人机验证后才能继续执行。` } -function monitorPlatformBlockedReason(platform: string): string | null { +function monitorPlatformAccountBlockedReason(platform: string): string | null { const ownedAccounts = state.accounts.filter( (item) => item.platform === platform && isAccountOwnedByCurrentClient(item), ) @@ -502,6 +504,13 @@ function monitorPlatformBlockedReason(platform: string): string | null { return firstBlockedReason } +function monitorPlatformBlockedReason(platform: string): string | null { + if (monitorPlatformCircuitBreakers.has(platform)) { + return `${runtimePlatformLabel(platform)} 已因账号人机验证暂停，完成验证并刷新账号状态后恢复。` + } + return monitorPlatformAccountBlockedReason(platform) +} + function blockedMonitorPlatforms(): Set { const blocked = new Set() for (const platform of aiPlatformCatalog.map((item) => item.id)) { @@ -883,6 +892,7 @@ function stopRuntime(): void { function clearRuntimeState(): void { initPublishScheduler() + monitorPlatformCircuitBreakers.clear() state.tasks.clear() state.accounts = [] state.accountProfiles.clear() @@ -1701,6 +1711,7 @@ export async function requestRuntimeAccountProbe( }) if (snapshot.authState === 'active') { clearMonitorAccountCooldown(account.id) + monitorPlatformCircuitBreakers.delete(account.platform) } enqueueAccountHealthReport(account.id) await mirrorProjectedAccountProfile(account.id) @@ -1771,6 +1782,7 @@ export function noteRuntimeAccountBound(account: DesktopAccountInfo): void { avatarUrl: normalizedAccount.avatar_url, }) clearMonitorAccountCooldown(normalizedAccount.id) + monitorPlatformCircuitBreakers.delete(normalizedAccount.platform) enqueueAccountHealthReport(normalizedAccount.id) refreshAccountNames() } @@ -3277,6 +3289,48 @@ function withMonitorAccountDiagnostics( } } +function isMonitorPlatformCircuitBreakerReason(reason: MonitorAccountFailoverReason): boolean { + return ( + reason === 'risk_control' || + reason === 'challenge_required' || + reason === 'authorization_failed' + ) +} + +function tripMonitorPlatformCircuitBreaker( + task: RuntimeTaskRecord, + reason: MonitorAccountFailoverReason, +): void { + monitorPlatformCircuitBreakers.add(task.platform) + const canceledTasks = cancelPendingMonitorTasksForPlatform(task.platform, task.id) + const summary = `${runtimePlatformLabel(task.platform)} 已无其他可用账号，后续 ${canceledTasks.length} 个本地任务已自动取消。` + + for (const canceledTask of canceledTasks) { + const existing = state.tasks.get(canceledTask.taskId) + if (!existing || state.activeExecutions.has(canceledTask.taskId)) { + continue + } + existing.status = 'aborted' + existing.summary = summary + existing.error = { + code: 'desktop_monitor_platform_challenge_canceled', + message: summary, + platform: task.platform, + failover_reason: reason, + } + existing.leaseToken = null + existing.updatedAt = Date.now() + state.tasks.set(existing.id, existing) + } + + recordActivity('warn', '模型后续任务已自动取消', summary) + syncSchedulerSurface() + emitRuntimeInvalidated('account-health', { + accountId: task.accountId, + message: summary, + }) +} + function monitorAccountPoolForTask(task: RuntimeTaskRecord): DesktopAccountInfo[] { const platformAccounts = state.accounts.filter((account) => account.platform === task.platform) const selected = selectMonitorAccountCandidates( @@ -3405,6 +3459,8 @@ async function executeMonitorTaskWithAccountPool( '模型账号自动切换', `${runtimePlatformLabel(task.platform)} 账号 ${account.display_name} 本轮返回 ${failoverReason}，${task.title} 将改用 ${nextAccount.display_name} 继续采集。`, ) + } else if (isMonitorPlatformCircuitBreakerReason(failoverReason)) { + tripMonitorPlatformCircuitBreaker(attemptTask, failoverReason) } }