feat(desktop): let manual collect-now bypass a same-day platform verification block

Scope the monitor platform circuit breaker to the business date that
tripped it, so it self-clears the next day, and route ordinary
account/platform blocks through an overrideable set. High-lane
collect-now tasks (manual trigger / priority >= 5000) may now force a
retry against a challenge_required account and bypass an overrideable
platform block for the same platform, while hard Playwright blocks and
expired/revoked accounts stay blocked. The lease pull plan probes the
high lane before regular work whenever a platform is challenge-blocked
or a high-priority signal is pending, and the client can request
specific lanes via the new LeaseDesktopTaskRequest.lanes field.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-22 23:56:21 +08:00
parent 184646f9fe
commit 7988a65018
8 changed files with 360 additions and 30 deletions
@@ -40,6 +40,36 @@ describe('monitor account pool', () => {
]) ])
}) })
it('allows an explicitly requested challenge retry without admitting expired accounts', () => {
const entries = [
{
id: 'challenge-retry',
authState: 'challenge_required',
hasLocalAuthorization: true,
allowForcedAttempt: true,
},
{ id: 'challenge-blocked', authState: 'challenge_required', hasLocalAuthorization: true },
{
id: 'expired',
authState: 'expired',
hasLocalAuthorization: true,
allowForcedAttempt: true,
},
{
id: 'expiring-retry',
authState: 'expiring_soon',
hasLocalAuthorization: true,
allowForcedAttempt: true,
},
]
coolDownMonitorAccount('challenge-retry', 'challenge_required', 1_000)
expect(selectMonitorAccountCandidates(entries, null, 1_001).map((item) => item.id)).toEqual([
'challenge-retry',
'expiring-retry',
])
})
it('releases a temporarily cooling account after its cooldown', () => { it('releases a temporarily cooling account after its cooldown', () => {
const entries = [{ id: 'account', authState: 'active', hasLocalAuthorization: true }] const entries = [{ id: 'account', authState: 'active', hasLocalAuthorization: true }]
coolDownMonitorAccount('account', 'adapter_failure', 10_000) coolDownMonitorAccount('account', 'adapter_failure', 10_000)
@@ -13,6 +13,7 @@ export type MonitorAccountPoolEntry = {
id: string id: string
authState: string authState: string
hasLocalAuthorization: boolean hasLocalAuthorization: boolean
allowForcedAttempt?: boolean
} }
type MonitorAccountCooldown = { type MonitorAccountCooldown = {
@@ -55,8 +56,9 @@ export function selectMonitorAccountCandidates(
.filter( .filter(
(entry) => (entry) =>
entry.hasLocalAuthorization && entry.hasLocalAuthorization &&
!isMonitorAccountAuthBlocked(entry.authState) && (!isMonitorAccountAuthBlocked(entry.authState) ||
(cooldowns.get(entry.id)?.until ?? 0) <= now, (entry.authState === 'challenge_required' && entry.allowForcedAttempt === true)) &&
((cooldowns.get(entry.id)?.until ?? 0) <= now || entry.allowForcedAttempt === true),
) )
.sort((left, right) => { .sort((left, right) => {
if (left.id === preferredAccountId && right.id !== preferredAccountId) { if (left.id === preferredAccountId && right.id !== preferredAccountId) {
@@ -0,0 +1,39 @@
import { beforeEach, describe, expect, it } from 'vitest'
import {
clearMonitorPlatformBreaker,
isMonitorPlatformBreakerActive,
resetMonitorPlatformBreakers,
tripMonitorPlatformBreaker,
} from './monitor-platform-breaker'
describe('monitor platform business-day breaker', () => {
beforeEach(() => resetMonitorPlatformBreakers())
it('blocks the challenged platform for the same business day', () => {
tripMonitorPlatformBreaker('doubao', '2026-07-21')
expect(isMonitorPlatformBreakerActive('doubao', '2026-07-21')).toBe(true)
expect(isMonitorPlatformBreakerActive('qwen', '2026-07-21')).toBe(false)
})
it('expires automatically on the next business day', () => {
tripMonitorPlatformBreaker('doubao', '2026-07-21')
expect(isMonitorPlatformBreakerActive('doubao', '2026-07-22')).toBe(false)
expect(isMonitorPlatformBreakerActive('doubao', '2026-07-22')).toBe(false)
})
it('only applies to the exact business day recorded by the failed task', () => {
tripMonitorPlatformBreaker('doubao', '2026-07-22')
expect(isMonitorPlatformBreakerActive('doubao', '2026-07-21')).toBe(false)
})
it('can be cleared immediately after a successful forced collection', () => {
tripMonitorPlatformBreaker('doubao', '2026-07-22')
clearMonitorPlatformBreaker('doubao')
expect(isMonitorPlatformBreakerActive('doubao', '2026-07-22')).toBe(false)
})
})
@@ -0,0 +1,35 @@
const blockedBusinessDates = new Map<string, string>()
export function tripMonitorPlatformBreaker(platform: string, businessDate: string): void {
const normalizedPlatform = platform.trim()
const normalizedBusinessDate = businessDate.trim()
if (!normalizedPlatform || !normalizedBusinessDate) {
return
}
blockedBusinessDates.set(normalizedPlatform, normalizedBusinessDate)
}
export function isMonitorPlatformBreakerActive(
platform: string,
currentBusinessDate: string,
): boolean {
const normalizedPlatform = platform.trim()
const normalizedCurrentBusinessDate = currentBusinessDate.trim()
const blockedBusinessDate = blockedBusinessDates.get(normalizedPlatform)
if (!blockedBusinessDate) {
return false
}
if (blockedBusinessDate !== normalizedCurrentBusinessDate) {
blockedBusinessDates.delete(normalizedPlatform)
return false
}
return true
}
export function clearMonitorPlatformBreaker(platform: string): void {
blockedBusinessDates.delete(platform.trim())
}
export function resetMonitorPlatformBreakers(): void {
blockedBusinessDates.clear()
}
@@ -10,6 +10,7 @@ vi.mock('electron/main', () => ({
})) }))
import { import {
buildMonitorLeasePullPlan,
cancelPendingMonitorTasksForPlatform, cancelPendingMonitorTasksForPlatform,
enqueueMonitorLeaseTask, enqueueMonitorLeaseTask,
getMonitorSchedulerSnapshot, getMonitorSchedulerSnapshot,
@@ -135,12 +136,86 @@ describe('monitor scheduler', () => {
).toBe('collect-now') ).toBe('collect-now')
}) })
it('lets collect-now high lane bypass a blocked platform while normal tasks remain blocked', () => {
enqueueMonitorLeaseTask({
taskId: 'blocked-normal',
clientId: 'client-1',
platform: 'doubao',
routing: 'rabbitmq-primary',
priority: 100,
lane: 'normal',
questionKey: 'q:normal',
})
enqueueMonitorLeaseTask({
taskId: 'collect-now',
clientId: 'client-1',
platform: 'doubao',
routing: 'rabbitmq-primary',
priority: 5000,
lane: 'high',
questionKey: 'q:collect-now',
})
expect(
selectNextMonitorTask({
maxConcurrency: 2,
activePlatforms: new Set(),
activeQuestionKeys: new Set(),
overrideableBlockedPlatforms: new Set(['doubao']),
activeCount: 0,
})?.taskId,
).toBe('collect-now')
})
it('does not let collect-now bypass a hard platform block', () => {
enqueueMonitorLeaseTask({
taskId: 'collect-now',
clientId: 'client-1',
platform: 'doubao',
routing: 'rabbitmq-primary',
priority: 5000,
lane: 'high',
questionKey: 'q:collect-now',
})
expect(
selectNextMonitorTask({
maxConcurrency: 2,
activePlatforms: new Set(),
activeQuestionKeys: new Set(),
blockedPlatforms: new Set(['doubao']),
activeCount: 0,
}),
).toBeNull()
})
it('probes high lane before regular work when a platform is challenge-blocked', () => {
expect(
buildMonitorLeasePullPlan({
highPrioritySignalPending: false,
hasOverrideableBlockedPlatforms: true,
hasRegularEligiblePlatforms: true,
}),
).toEqual(['high', 'regular'])
})
it('uses the regular ordered lease when no challenge block or signal exists', () => {
expect(
buildMonitorLeasePullPlan({
highPrioritySignalPending: false,
hasOverrideableBlockedPlatforms: false,
hasRegularEligiblePlatforms: true,
}),
).toEqual(['regular'])
})
it('removes later same-platform tasks after the only account hits a challenge', () => { it('removes later same-platform tasks after the only account hits a challenge', () => {
enqueueMonitorLeaseTask({ enqueueMonitorLeaseTask({
taskId: 'doubao-current', taskId: 'doubao-current',
clientId: 'client-1', clientId: 'client-1',
platform: 'doubao', platform: 'doubao',
routing: 'rabbitmq-primary', routing: 'rabbitmq-primary',
businessDate: '2026-06-22',
questionKey: 'q:current', questionKey: 'q:current',
}) })
enqueueMonitorLeaseTask({ enqueueMonitorLeaseTask({
@@ -148,8 +223,17 @@ describe('monitor scheduler', () => {
clientId: 'client-1', clientId: 'client-1',
platform: 'doubao', platform: 'doubao',
routing: 'rabbitmq-primary', routing: 'rabbitmq-primary',
businessDate: '2026-06-22',
questionKey: 'q:next', questionKey: 'q:next',
}) })
enqueueMonitorLeaseTask({
taskId: 'doubao-tomorrow',
clientId: 'client-1',
platform: 'doubao',
routing: 'rabbitmq-primary',
businessDate: '2026-06-23',
questionKey: 'q:tomorrow',
})
enqueueMonitorLeaseTask({ enqueueMonitorLeaseTask({
taskId: 'qwen-next', taskId: 'qwen-next',
clientId: 'client-1', clientId: 'client-1',
@@ -159,14 +243,14 @@ describe('monitor scheduler', () => {
}) })
noteMonitorTaskActivated('doubao-current') noteMonitorTaskActivated('doubao-current')
expect(cancelPendingMonitorTasksForPlatform('doubao', 'doubao-current')).toEqual([ expect(cancelPendingMonitorTasksForPlatform('doubao', 'doubao-current', '2026-06-22')).toEqual([
expect.objectContaining({ taskId: 'doubao-next', platform: 'doubao' }), expect.objectContaining({ taskId: 'doubao-next', platform: 'doubao' }),
]) ])
expect( expect(
getMonitorSchedulerSnapshot() getMonitorSchedulerSnapshot()
.tasks.map((task) => task.taskId) .tasks.map((task) => task.taskId)
.sort(), .sort(),
).toEqual(['doubao-current', 'qwen-next']) ).toEqual(['doubao-current', 'doubao-tomorrow', 'qwen-next'])
expect( expect(
selectNextMonitorTask({ selectNextMonitorTask({
maxConcurrency: 2, maxConcurrency: 2,
@@ -65,9 +65,27 @@ interface MonitorSchedulerSelectionOptions {
activePlatforms: ReadonlySet<string> activePlatforms: ReadonlySet<string>
activeQuestionKeys: ReadonlySet<string> activeQuestionKeys: ReadonlySet<string>
blockedPlatforms?: ReadonlySet<string> blockedPlatforms?: ReadonlySet<string>
overrideableBlockedPlatforms?: ReadonlySet<string>
activeCount: number activeCount: number
} }
export type MonitorLeasePullStep = 'high' | 'regular'
export function buildMonitorLeasePullPlan(input: {
highPrioritySignalPending: boolean
hasOverrideableBlockedPlatforms: boolean
hasRegularEligiblePlatforms: boolean
}): MonitorLeasePullStep[] {
const plan: MonitorLeasePullStep[] = []
if (input.highPrioritySignalPending || input.hasOverrideableBlockedPlatforms) {
plan.push('high')
}
if (input.hasRegularEligiblePlatforms) {
plan.push('regular')
}
return plan
}
interface MonitorTaskMetadata { interface MonitorTaskMetadata {
title: string | null title: string | null
businessDate: string | null businessDate: string | null
@@ -477,8 +495,10 @@ export function noteMonitorTaskCanceled(taskId: string): void {
export function cancelPendingMonitorTasksForPlatform( export function cancelPendingMonitorTasksForPlatform(
platform: string, platform: string,
currentTaskId?: string | null, currentTaskId?: string | null,
businessDate?: string | null,
): MonitorSchedulerTaskRecord[] { ): MonitorSchedulerTaskRecord[] {
const normalizedPlatform = platform.trim() const normalizedPlatform = platform.trim()
const normalizedBusinessDate = businessDate?.trim() || null
if (!normalizedPlatform) { if (!normalizedPlatform) {
return [] return []
} }
@@ -489,7 +509,8 @@ export function cancelPendingMonitorTasksForPlatform(
if ( if (
task.platform !== normalizedPlatform || task.platform !== normalizedPlatform ||
taskId === currentTaskId || taskId === currentTaskId ||
task.state === 'active' task.state === 'active' ||
(normalizedBusinessDate !== null && task.businessDate !== normalizedBusinessDate)
) { ) {
continue continue
} }
@@ -531,15 +552,19 @@ export function selectNextMonitorTask(
}) })
for (const candidate of candidates) { for (const candidate of candidates) {
const isCollectNow = candidate.lane === 'high'
if (options.blockedPlatforms?.has(candidate.platform)) { if (options.blockedPlatforms?.has(candidate.platform)) {
continue continue
} }
if (options.overrideableBlockedPlatforms?.has(candidate.platform) && !isCollectNow) {
continue
}
if (options.activePlatforms.has(candidate.platform)) { if (options.activePlatforms.has(candidate.platform)) {
continue continue
} }
const bypassQuestionCooldown = candidate.lane === 'high' const bypassQuestionCooldown = isCollectNow
const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0 const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0
if (platformCooldown > now) { if (platformCooldown > now) {
@@ -30,6 +30,7 @@ import {
getAccountHealthSnapshot, getAccountHealthSnapshot,
getProjectedAccountHealth, getProjectedAccountHealth,
markTrackedAccountBound, markTrackedAccountBound,
markTrackedAccountExecutionSucceeded,
markTrackedAccountMissingPartition, markTrackedAccountMissingPartition,
probeTrackedAccount, probeTrackedAccount,
reportAccountFailure, reportAccountFailure,
@@ -71,6 +72,13 @@ import {
type MonitorAccountFailoverReason, type MonitorAccountFailoverReason,
} from './monitor-account-pool' } from './monitor-account-pool'
import { import {
clearMonitorPlatformBreaker,
isMonitorPlatformBreakerActive,
resetMonitorPlatformBreakers,
tripMonitorPlatformBreaker,
} from './monitor-platform-breaker'
import {
buildMonitorLeasePullPlan,
cancelPendingMonitorTasksForPlatform, cancelPendingMonitorTasksForPlatform,
enqueueMonitorLeaseTask, enqueueMonitorLeaseTask,
getMonitorSchedulerSnapshot, getMonitorSchedulerSnapshot,
@@ -192,8 +200,6 @@ const accountHealthReportDebounceMs = 1_000
const accountHealthReportMaxBatchSize = 100 const accountHealthReportMaxBatchSize = 100
const maxPlaywrightCDPSoftRecoveries = 3 const maxPlaywrightCDPSoftRecoveries = 3
const playwrightCDPFatalMarker = '--geo-cdp-fatal' const playwrightCDPFatalMarker = '--geo-cdp-fatal'
const monitorPlatformCircuitBreakers = new Set<string>()
interface RuntimeTaskRecord { interface RuntimeTaskRecord {
id: string id: string
jobId: string jobId: string
@@ -286,6 +292,7 @@ interface RuntimeState {
leaseInFlightKinds: Set<'publish' | 'monitor'> leaseInFlightKinds: Set<'publish' | 'monitor'>
publishFallbackBackoffUntil: number publishFallbackBackoffUntil: number
monitorFallbackBackoffUntil: number monitorFallbackBackoffUntil: number
monitorHighPriorityPullRequested: boolean
heartbeatTimer: ReturnType<typeof setInterval> | null heartbeatTimer: ReturnType<typeof setInterval> | null
pullTimer: ReturnType<typeof setInterval> | null pullTimer: ReturnType<typeof setInterval> | null
accountSyncTimer: ReturnType<typeof setInterval> | null accountSyncTimer: ReturnType<typeof setInterval> | null
@@ -320,6 +327,7 @@ const state: RuntimeState = {
leaseInFlightKinds: new Set<'publish' | 'monitor'>(), leaseInFlightKinds: new Set<'publish' | 'monitor'>(),
publishFallbackBackoffUntil: 0, publishFallbackBackoffUntil: 0,
monitorFallbackBackoffUntil: 0, monitorFallbackBackoffUntil: 0,
monitorHighPriorityPullRequested: false,
heartbeatTimer: null, heartbeatTimer: null,
pullTimer: null, pullTimer: null,
accountSyncTimer: null, accountSyncTimer: null,
@@ -473,6 +481,26 @@ function accountActionRequiredSummary(
return `${label} 账号需要完成人机验证后才能继续执行。` return `${label} 账号需要完成人机验证后才能继续执行。`
} }
function accountChallengeBusinessDate(input: {
lastAuthFailureAt: number | null
lastVerifiedAt: number | null
}): string | null {
const timestamp = input.lastAuthFailureAt ?? input.lastVerifiedAt
return timestamp ? currentMonitoringBusinessDate(new Date(timestamp)) : null
}
function isPriorBusinessDayChallenge(input: {
authState: string
lastAuthFailureAt: number | null
lastVerifiedAt: number | null
}): boolean {
if (input.authState !== 'challenge_required') {
return false
}
const challengeDate = accountChallengeBusinessDate(input)
return Boolean(challengeDate && challengeDate < currentMonitoringBusinessDate())
}
function monitorPlatformAccountBlockedReason(platform: string): string | null { function monitorPlatformAccountBlockedReason(platform: string): string | null {
const ownedAccounts = state.accounts.filter( const ownedAccounts = state.accounts.filter(
(item) => item.platform === platform && isAccountOwnedByCurrentClient(item), (item) => item.platform === platform && isAccountOwnedByCurrentClient(item),
@@ -496,6 +524,9 @@ function monitorPlatformAccountBlockedReason(platform: string): string | null {
if (!isMonitorAccountAuthBlocked(projected.authState)) { if (!isMonitorAccountAuthBlocked(projected.authState)) {
return null return null
} }
if (isPriorBusinessDayChallenge(projected)) {
return null
}
firstBlockedReason ??= firstBlockedReason ??=
projected.authState === 'challenge_required' projected.authState === 'challenge_required'
? accountActionRequiredSummary(account.platform, projected.authReason) ? accountActionRequiredSummary(account.platform, projected.authReason)
@@ -505,7 +536,7 @@ function monitorPlatformAccountBlockedReason(platform: string): string | null {
} }
function monitorPlatformBlockedReason(platform: string): string | null { function monitorPlatformBlockedReason(platform: string): string | null {
if (monitorPlatformCircuitBreakers.has(platform)) { if (isMonitorPlatformBreakerActive(platform, currentMonitoringBusinessDate())) {
return `${runtimePlatformLabel(platform)} 已因账号人机验证暂停,完成验证并刷新账号状态后恢复。` return `${runtimePlatformLabel(platform)} 已因账号人机验证暂停,完成验证并刷新账号状态后恢复。`
} }
return monitorPlatformAccountBlockedReason(platform) return monitorPlatformAccountBlockedReason(platform)
@@ -892,7 +923,7 @@ function stopRuntime(): void {
function clearRuntimeState(): void { function clearRuntimeState(): void {
initPublishScheduler() initPublishScheduler()
monitorPlatformCircuitBreakers.clear() resetMonitorPlatformBreakers()
state.tasks.clear() state.tasks.clear()
state.accounts = [] state.accounts = []
state.accountProfiles.clear() state.accountProfiles.clear()
@@ -910,6 +941,7 @@ function clearRuntimeState(): void {
state.lastError = null state.lastError = null
state.publishFallbackBackoffUntil = 0 state.publishFallbackBackoffUntil = 0
state.monitorFallbackBackoffUntil = 0 state.monitorFallbackBackoffUntil = 0
state.monitorHighPriorityPullRequested = false
setSchedulerQueueDepth(localQueueDepth()) setSchedulerQueueDepth(localQueueDepth())
} }
@@ -1124,6 +1156,7 @@ async function handleMonitoringDispatchSignal(event: DesktopTaskEventMessage): P
} }
if ((event.lane ?? '').trim() === 'high') { if ((event.lane ?? '').trim() === 'high') {
state.monitorHighPriorityPullRequested = true
recordActivity( recordActivity(
'info', 'info',
'收到监控高优先级信号', '收到监控高优先级信号',
@@ -1711,7 +1744,7 @@ export async function requestRuntimeAccountProbe(
}) })
if (snapshot.authState === 'active') { if (snapshot.authState === 'active') {
clearMonitorAccountCooldown(account.id) clearMonitorAccountCooldown(account.id)
monitorPlatformCircuitBreakers.delete(account.platform) clearMonitorPlatformBreaker(account.platform)
} }
enqueueAccountHealthReport(account.id) enqueueAccountHealthReport(account.id)
await mirrorProjectedAccountProfile(account.id) await mirrorProjectedAccountProfile(account.id)
@@ -1782,7 +1815,7 @@ export function noteRuntimeAccountBound(account: DesktopAccountInfo): void {
avatarUrl: normalizedAccount.avatar_url, avatarUrl: normalizedAccount.avatar_url,
}) })
clearMonitorAccountCooldown(normalizedAccount.id) clearMonitorAccountCooldown(normalizedAccount.id)
monitorPlatformCircuitBreakers.delete(normalizedAccount.platform) clearMonitorPlatformBreaker(normalizedAccount.platform)
enqueueAccountHealthReport(normalizedAccount.id) enqueueAccountHealthReport(normalizedAccount.id)
refreshAccountNames() refreshAccountNames()
} }
@@ -1991,15 +2024,12 @@ function pumpExecutionLoop(): void {
} }
if (canStartAnotherMonitorExecution()) { if (canStartAnotherMonitorExecution()) {
const blockedMonitorExecutionPlatforms = new Set([
...blockedPlaywrightPlatforms,
...blockedMonitorPlatforms(),
])
const selected = selectNextMonitorTask({ const selected = selectNextMonitorTask({
maxConcurrency: resolveAdaptiveMonitorConcurrency(), maxConcurrency: resolveAdaptiveMonitorConcurrency(),
activePlatforms: activeMonitorPlatforms(), activePlatforms: activeMonitorPlatforms(),
activeQuestionKeys: activeMonitorQuestionKeys(), activeQuestionKeys: activeMonitorQuestionKeys(),
blockedPlatforms: blockedMonitorExecutionPlatforms, blockedPlatforms: blockedPlaywrightPlatforms,
overrideableBlockedPlatforms: blockedMonitorPlatforms(),
activeCount: activeMonitorCount(), activeCount: activeMonitorCount(),
}) })
if (selected) { if (selected) {
@@ -2086,6 +2116,55 @@ async function leaseSpecificTask(
} }
} }
function monitorLeasePlatformIDs(includeChallengeBlocked: boolean): string[] {
const hardBlocked = blockedPlaywrightExecutionPlatforms()
const platformIDs = includeChallengeBlocked
? aiPlatformCatalog.map((platform) => platform.id)
: eligibleMonitorPlatformIds()
return platformIDs.filter((platform) => !hardBlocked.has(platform))
}
async function leaseNextMonitorDesktopTask(): Promise<LeaseDesktopTaskResponse> {
const allPlatformIDs = monitorLeasePlatformIDs(true)
const eligiblePlatformIDs = monitorLeasePlatformIDs(false)
const hasOverrideableBlockedPlatforms = blockedMonitorPlatforms().size > 0
const leaseHighPriority = async (): Promise<LeaseDesktopTaskResponse> => {
if (allPlatformIDs.length === 0) {
return { task: null }
}
return await leaseDesktopTask({
kind: 'monitor',
platform_ids: allPlatformIDs,
lanes: ['high'],
})
}
const plan = buildMonitorLeasePullPlan({
highPrioritySignalPending: state.monitorHighPriorityPullRequested,
hasOverrideableBlockedPlatforms,
hasRegularEligiblePlatforms: eligiblePlatformIDs.length > 0,
})
for (const step of plan) {
if (step === 'high') {
const highPriority = await leaseHighPriority()
if (highPriority.task) {
return highPriority
}
state.monitorHighPriorityPullRequested = false
continue
}
const regular = await leaseDesktopTask({
kind: 'monitor',
platform_ids: eligiblePlatformIDs,
})
if (regular.task) {
return regular
}
}
return { task: null }
}
async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> { async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> {
if (kind === 'monitor') { if (kind === 'monitor') {
if (!canRunLive(state.session) || isLeaseInFlight('monitor') || playwrightCDPFatal) { if (!canRunLive(state.session) || isLeaseInFlight('monitor') || playwrightCDPFatal) {
@@ -2098,15 +2177,7 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> {
state.leaseInFlightKinds.add('monitor') state.leaseInFlightKinds.add('monitor')
try { try {
const eligiblePlatformIds = eligibleMonitorPlatformIds() const leased = await leaseNextMonitorDesktopTask()
if (eligiblePlatformIds.length === 0) {
state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs
return
}
const leased = await leaseDesktopTask({
kind: 'monitor',
platform_ids: eligiblePlatformIds,
})
state.lastPullAt = Date.now() state.lastPullAt = Date.now()
state.lastPullStatus = 'success' state.lastPullStatus = 'success'
noteTransportPull(true) noteTransportPull(true)
@@ -3301,8 +3372,10 @@ function tripMonitorPlatformCircuitBreaker(
task: RuntimeTaskRecord, task: RuntimeTaskRecord,
reason: MonitorAccountFailoverReason, reason: MonitorAccountFailoverReason,
): void { ): void {
monitorPlatformCircuitBreakers.add(task.platform) const businessDate =
const canceledTasks = cancelPendingMonitorTasksForPlatform(task.platform, task.id) resolveMonitoringBusinessDate(task.payload) ?? currentMonitoringBusinessDate()
tripMonitorPlatformBreaker(task.platform, businessDate)
const canceledTasks = cancelPendingMonitorTasksForPlatform(task.platform, task.id, businessDate)
const summary = `${runtimePlatformLabel(task.platform)} 已无其他可用账号,后续 ${canceledTasks.length} 个本地任务已自动取消。` const summary = `${runtimePlatformLabel(task.platform)} 已无其他可用账号,后续 ${canceledTasks.length} 个本地任务已自动取消。`
for (const canceledTask of canceledTasks) { for (const canceledTask of canceledTasks) {
@@ -3331,6 +3404,34 @@ function tripMonitorPlatformCircuitBreaker(
}) })
} }
function isCollectNowMonitorTask(task: RuntimeTaskRecord): boolean {
return (
task.payload.lane === 'high' ||
task.payload.trigger_source === 'manual' ||
(typeof task.payload.priority === 'number' && task.payload.priority >= 5000)
)
}
function shouldForceMonitorAccountAttempt(
task: RuntimeTaskRecord,
health: {
authState: string
lastAuthFailureAt: number | null
lastVerifiedAt: number | null
},
): boolean {
if (isCollectNowMonitorTask(task)) {
return health.authState !== 'expired' && health.authState !== 'revoked'
}
if (health.authState !== 'challenge_required') {
return false
}
const challengeDate = accountChallengeBusinessDate(health)
const taskBusinessDate =
resolveMonitoringBusinessDate(task.payload) ?? currentMonitoringBusinessDate()
return Boolean(challengeDate && challengeDate < taskBusinessDate)
}
function monitorAccountPoolForTask(task: RuntimeTaskRecord): DesktopAccountInfo[] { function monitorAccountPoolForTask(task: RuntimeTaskRecord): DesktopAccountInfo[] {
const platformAccounts = state.accounts.filter((account) => account.platform === task.platform) const platformAccounts = state.accounts.filter((account) => account.platform === task.platform)
const selected = selectMonitorAccountCandidates( const selected = selectMonitorAccountCandidates(
@@ -3345,6 +3446,7 @@ function monitorAccountPoolForTask(task: RuntimeTaskRecord): DesktopAccountInfo[
id: account.id, id: account.id,
authState: projected.authState, authState: projected.authState,
hasLocalAuthorization: hasLocalAccountAuthorization(account), hasLocalAuthorization: hasLocalAccountAuthorization(account),
allowForcedAttempt: shouldForceMonitorAccountAttempt(task, projected),
} }
}), }),
task.accountId || null, task.accountId || null,
@@ -3419,8 +3521,15 @@ async function executeMonitorTaskWithAccountPool(
} }
const previousSnapshot = getAccountHealthSnapshot(identity.id) const previousSnapshot = getAccountHealthSnapshot(identity.id)
const readiness = await ensureAccountReady(identity) const projected = getProjectedAccountHealth({
if (readiness.authState !== 'active') { accountId: account.id,
platform: account.platform,
health: account.health,
verifiedAt: account.verified_at,
})
const forceAccountAttempt = shouldForceMonitorAccountAttempt(attemptTask, projected)
const readiness = forceAccountAttempt ? null : await ensureAccountReady(identity)
if (readiness && readiness.authState !== 'active') {
lastResult = buildAccountBlockedResult(attemptTask, readiness) lastResult = buildAccountBlockedResult(attemptTask, readiness)
maybeRecordAccountAccessAlert(attemptTask, identity, previousSnapshot, readiness) maybeRecordAccountAccessAlert(attemptTask, identity, previousSnapshot, readiness)
enqueueAccountHealthReport(identity.id) enqueueAccountHealthReport(identity.id)
@@ -3438,6 +3547,11 @@ async function executeMonitorTaskWithAccountPool(
} }
} }
await maybeReportTaskAuthFailure(attemptTask, lastResult, identity) await maybeReportTaskAuthFailure(attemptTask, lastResult, identity)
if (lastResult.status === 'succeeded') {
markTrackedAccountExecutionSucceeded(identity)
clearMonitorPlatformBreaker(task.platform)
enqueueAccountHealthReport(identity.id)
}
} }
const failoverReason = classifyMonitorAccountFailover(lastResult) const failoverReason = classifyMonitorAccountFailover(lastResult)
+1
View File
@@ -346,6 +346,7 @@ export interface LeaseDesktopTaskRequest {
task_id?: string task_id?: string
kind?: 'publish' | 'monitor' kind?: 'publish' | 'monitor'
platform_ids?: string[] platform_ids?: string[]
lanes?: string[]
} }
export interface LeaseDesktopTaskResponse { export interface LeaseDesktopTaskResponse {