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) <noreply@anthropic.com>
2026-07-12 14:17:27 +08:00
parent 70b9f75aea
commit 2706c07381
4 changed files with 501 additions and 23 deletions
@@ -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()
+  })
+})
@@ -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<string, number>()
+const cooldowns = new Map<string, MonitorAccountCooldown>()
+
+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,
+      })),
+  }
+}
@@ -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<string> {
@@ -574,6 +587,9 @@ function withAuthorizationRetryPolicy(
 async function resolveAuthorizationPreflightBlock(
  task: RuntimeTaskRecord,
 ): Promise<AdapterExecutionResult | null> {
+  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<AdapterExecutionResult> {
  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<string, JsonValue>,
+  signal: AbortSignal,
+): Promise<AdapterExecutionResult> {
+  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<string, JsonValue>,
+  signal: AbortSignal,
+): Promise<AdapterExecutionResult> {
+  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()
@@ -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: {