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) <noreply@anthropic.com>
This commit is contained in:
@@ -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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<string>()
|
||||
|
||||
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<string> {
|
||||
const blocked = new Set<string>()
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user