fix: lease monitor tasks by account identity

apps/desktop-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@geo/desktop-client",
-  "version": "0.1.2",
+  "version": "0.1.3",
   "private": true,
   "description": "省心推客户端 — 连接和管理您的媒体账号，轻松发布和监控内容表现。",
   "author": {

apps/desktop-client/src/main/monitor-scheduler.test.ts

@@ -0,0 +1,135 @@
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('electron/main', () => ({
+  app: {
+    getPath: () => join(tmpdir(), 'geo-rankly-monitor-scheduler-test'),
+  },
+}))
+
+import {
+  enqueueMonitorLeaseTask,
+  initMonitorScheduler,
+  noteMonitorTaskActivated,
+  noteMonitorTaskCompleted,
+  selectNextMonitorTask,
+} from './monitor-scheduler'
+
+describe('monitor scheduler', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-06-22T00:00:00.000Z'))
+    initMonitorScheduler()
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+    initMonitorScheduler()
+  })
+
+  it('does not let high-priority monitor tasks bypass same-platform cooldown', () => {
+    enqueueMonitorLeaseTask({
+      taskId: 'first',
+      clientId: 'client-1',
+      platform: 'doubao',
+      routing: 'rabbitmq-primary',
+      questionKey: 'q:first',
+    })
+    expect(
+      selectNextMonitorTask({
+        maxConcurrency: 2,
+        activePlatforms: new Set(),
+        activeQuestionKeys: new Set(),
+        activeCount: 0,
+      })?.taskId,
+    ).toBe('first')
+    noteMonitorTaskActivated('first')
+    noteMonitorTaskCompleted('first')
+
+    enqueueMonitorLeaseTask({
+      taskId: 'urgent',
+      clientId: 'client-1',
+      platform: 'doubao',
+      routing: 'rabbitmq-primary',
+      priority: 5000,
+      lane: 'high',
+      questionKey: 'q:urgent',
+    })
+
+    expect(
+      selectNextMonitorTask({
+        maxConcurrency: 2,
+        activePlatforms: new Set(),
+        activeQuestionKeys: new Set(),
+        activeCount: 0,
+      }),
+    ).toBeNull()
+
+    vi.advanceTimersByTime(5_001)
+
+    expect(
+      selectNextMonitorTask({
+        maxConcurrency: 2,
+        activePlatforms: new Set(),
+        activeQuestionKeys: new Set(),
+        activeCount: 0,
+      })?.taskId,
+    ).toBe('urgent')
+  })
+
+  it('does not let high-priority monitor tasks bypass an active same-platform task', () => {
+    enqueueMonitorLeaseTask({
+      taskId: 'urgent-doubao',
+      clientId: 'client-1',
+      platform: 'doubao',
+      routing: 'rabbitmq-primary',
+      priority: 5000,
+      lane: 'high',
+      questionKey: 'q:urgent',
+    })
+
+    expect(
+      selectNextMonitorTask({
+        maxConcurrency: 2,
+        activePlatforms: new Set(['doubao']),
+        activeQuestionKeys: new Set(),
+        activeCount: 1,
+      }),
+    ).toBeNull()
+  })
+
+  it('prioritizes collect-now high lane before older normal tasks when platform is available', () => {
+    enqueueMonitorLeaseTask({
+      taskId: 'old-normal',
+      clientId: 'client-1',
+      platform: 'qwen',
+      routing: 'rabbitmq-primary',
+      priority: 100,
+      lane: 'normal',
+      questionKey: 'q:normal',
+    })
+
+    vi.advanceTimersByTime(1_000)
+
+    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(),
+        activeCount: 0,
+      })?.taskId,
+    ).toBe('collect-now')
+  })
+})

apps/desktop-client/src/main/monitor-scheduler.ts

@@ -74,7 +74,7 @@ interface MonitorTaskMetadata {
   questionText: string | null
 }
 
-const schedulerStateVersion = 5
+const schedulerStateVersion = 6
 const schedulerQuestionCooldownMs = 45_000
 const schedulerPlatformCooldownMs = 5_000
 const schedulerRestartRecoveryDelayMs = 90_000
@@ -492,10 +492,10 @@ export function selectNextMonitorTask(
       continue
     }
 
-    const bypassCooldowns = candidate.lane === 'high'
+    const bypassQuestionCooldown = candidate.lane === 'high'
 
     const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0
-    if (!bypassCooldowns && platformCooldown > now) {
+    if (platformCooldown > now) {
       continue
     }
 
@@ -508,7 +508,7 @@ export function selectNextMonitorTask(
         continue
       }
       const questionCooldown = draft.questionCooldowns[candidate.questionKey] ?? 0
-      if (!bypassCooldowns && questionCooldown > now) {
+      if (!bypassQuestionCooldown && questionCooldown > now) {
         continue
       }
     }
@@ -553,12 +553,12 @@ export function getMonitorSchedulerSnapshot(): MonitorSchedulerSnapshot {
 
 function leaseMissBackoffMs(leaseMisses: number): number {
   if (leaseMisses <= 1) {
-    return 20_000
+    return 15_000
   }
   if (leaseMisses === 2) {
-    return 60_000
+    return 30_000
   }
-  return 3 * 60_000
+  return 60_000
 }
 
 function metadataFromEvent(event: DesktopTaskEventMessage): MonitorTaskMetadata {

apps/desktop-client/src/main/runtime-controller.ts

@@ -63,13 +63,11 @@ import {
 } from './lease-manager'
 import {
   enqueueMonitorLeaseTask,
-  enqueueMonitorTaskFromEvent,
   getMonitorSchedulerSnapshot,
   initMonitorScheduler,
   noteMonitorTaskActivated,
   noteMonitorTaskCanceled,
   noteMonitorTaskCompleted,
-  noteMonitorTaskLeaseMiss,
   noteMonitorTaskLeased,
   selectNextMonitorTask,
 } from './monitor-scheduler'
@@ -149,6 +147,7 @@ type MonitorExecutionPhase =
 const heartbeatIntervalMs = 15_000
 const pullIntervalMs = 60_000
 const publishPullIntervalMs = 30_000
+const monitorPullFallbackIntervalMs = 10_000
 const accountSyncIntervalMs = 30_000
 const pumpWatchdogIntervalMs = 30_000
 const leaseExtendIntervalMs = 60_000
@@ -256,6 +255,7 @@ interface RuntimeState {
   activeExecutions: Map<string, ActiveRuntimeExecution>
   leaseInFlightKinds: Set<'publish' | 'monitor'>
   publishFallbackBackoffUntil: number
+  monitorFallbackBackoffUntil: number
   heartbeatTimer: ReturnType<typeof setInterval> | null
   pullTimer: ReturnType<typeof setInterval> | null
   accountSyncTimer: ReturnType<typeof setInterval> | null
@@ -289,6 +289,7 @@ const state: RuntimeState = {
   activeExecutions: new Map<string, ActiveRuntimeExecution>(),
   leaseInFlightKinds: new Set<'publish' | 'monitor'>(),
   publishFallbackBackoffUntil: 0,
+  monitorFallbackBackoffUntil: 0,
   heartbeatTimer: null,
   pullTimer: null,
   accountSyncTimer: null,
@@ -806,6 +807,7 @@ function clearRuntimeState(): void {
   state.onlineClientCount = null
   state.lastError = null
   state.publishFallbackBackoffUntil = 0
+  state.monitorFallbackBackoffUntil = 0
   setSchedulerQueueDepth(localQueueDepth())
 }
 
@@ -826,11 +828,11 @@ function schedulePullLoop(): void {
     clearInterval(state.pullTimer)
   }
 
-  setSchedulerNextPull(Date.now() + publishPullIntervalMs)
+  setSchedulerNextPull(Date.now() + monitorPullFallbackIntervalMs)
   state.pullTimer = setInterval(() => {
-    setSchedulerNextPull(Date.now() + publishPullIntervalMs)
+    setSchedulerNextPull(Date.now() + monitorPullFallbackIntervalMs)
     pumpExecutionLoop()
-  }, publishPullIntervalMs)
+  }, monitorPullFallbackIntervalMs)
 }
 
 function scheduleAccountSyncLoop(): void {
@@ -986,7 +988,7 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void {
 
   if (event.kind === 'monitor') {
     if (event.type === 'task_available') {
-      enqueueMonitorTaskFromEvent(event, 'rabbitmq-primary')
+      state.monitorFallbackBackoffUntil = 0
     }
     if (
       event.type === 'task_completed' ||
@@ -1027,13 +1029,8 @@ async function handleMonitoringDispatchSignal(event: DesktopTaskEventMessage): P
     )
   }
 
-  if (event.task_id.trim()) {
-    enqueueMonitorTaskFromEvent(event, 'rabbitmq-primary')
+  state.monitorFallbackBackoffUntil = 0
   pumpExecutionLoop()
-    return
-  }
-
-  await pullNextTask('monitor')
 }
 
 async function handleMonitoringTaskControl(event: DesktopTaskEventMessage): Promise<void> {
@@ -1889,11 +1886,21 @@ function pumpExecutionLoop(): void {
       activeCount: activeMonitorCount(),
     })
     if (selected) {
+      if (selected.source === 'monitoring_lease') {
         void leaseSpecificTask(selected, 'monitor')
+      } else {
+        noteMonitorTaskCanceled(selected.taskId)
+        void pullNextTask('monitor')
+      }
       return
     }
   }
 
+  if (shouldPullMonitorFallback()) {
+    void pullNextTask('monitor')
+    return
+  }
+
   return
 }
 
@@ -1917,7 +1924,7 @@ async function leaseSpecificTask(
     const leased = await leaseDesktopTask({ task_id: request.taskId })
     if (!leased.task) {
       if (expectedKind === 'monitor') {
-        noteMonitorTaskLeaseMiss(request.taskId)
+        noteMonitorTaskCanceled(request.taskId)
       } else {
         notePublishTaskLeaseMiss(request.taskId)
       }
@@ -1938,7 +1945,7 @@ async function leaseSpecificTask(
           state.tasks.set(request.taskId, existing)
         }
       } else {
-        noteMonitorTaskLeaseMiss(request.taskId)
+        noteMonitorTaskCanceled(request.taskId)
       }
     } else {
       notePublishTaskLeaseMiss(request.taskId)
@@ -1973,14 +1980,17 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> {
       noteSchedulerPull()
 
       if (leased.task) {
+        state.monitorFallbackBackoffUntil = 0
         state.leaseInFlightKinds.delete('monitor')
         await executeLeasedTask(leased, 'rabbitmq-primary')
         return
       }
+      state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs
     } catch (error) {
       state.lastPullAt = Date.now()
       state.lastPullStatus = 'failed'
       state.lastError = errorMessage(error)
+      state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs
       noteTransportPull(false)
       setSchedulerError(state.lastError)
     } finally {
@@ -2052,10 +2062,12 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> {
     noteSchedulerPull()
 
     enqueueLeasedMonitoringTasks(leased.tasks, routing)
+    state.monitorFallbackBackoffUntil = leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs
   } catch (error) {
     state.lastPullAt = Date.now()
     state.lastPullStatus = 'failed'
     state.lastError = errorMessage(error)
+    state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs
     noteTransportPull(false)
     setSchedulerError(state.lastError)
 
@@ -2176,7 +2188,7 @@ async function executeLeasedMonitoringTask(
 ): Promise<void> {
   const taskRecord = state.tasks.get(taskId)
   if (!taskRecord || taskRecord.kind !== 'monitor' || !taskRecord.leaseToken) {
-    noteMonitorTaskLeaseMiss(taskId)
+    noteMonitorTaskCanceled(taskId)
     return
   }
 
@@ -3609,6 +3621,14 @@ function shouldPullPublishFallback(): boolean {
   )
 }
 
+function shouldPullMonitorFallback(): boolean {
+  return (
+    Date.now() >= state.monitorFallbackBackoffUntil &&
+    !hasPublishBacklog() &&
+    canStartAnotherMonitorExecution()
+  )
+}
+
 function canStartAnotherMonitorExecution(): boolean {
   if (hasPublishBacklog() || isLeaseInFlight('publish') || isLeaseInFlight('monitor')) {
     return false

server/internal/tenant/app/desktop_task_service.go

@@ -28,6 +28,9 @@ import (
 const (
 	desktopPublishMaxAttempts                     = 3
 	maxConcurrentMonitorPlatformsPerDesktopClient = 2
+	monitorLeaseCandidateScanLimitSQL             = "16"
+	monitorSucceededTaskCooldownSQL               = "2 seconds"
+	monitorUnhealthyTaskCooldownSQL               = "30 seconds"
 )
 
 var errDesktopTaskLeaseDeferred = errors.New("desktop task lease deferred")
@@ -517,12 +520,6 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 	client *repository.DesktopClient,
 	params repository.DesktopTaskLeaseParams,
 ) (*repository.DesktopTask, error) {
-	accountIDs, err := s.loadMonitorLeaseAccountIDs(ctx, client)
-	if err != nil {
-		return nil, err
-	}
-	hasAccountIDs := len(accountIDs) > 0
-
 	tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
 	if err != nil {
 		return nil, err
@@ -533,8 +530,6 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 		client.TenantID,
 		client.WorkspaceID,
 		client.ID,
-		hasAccountIDs,
-		accountIDs,
 		params.AttemptID,
 		params.LeaseTokenHash,
 		maxConcurrentMonitorPlatformsPerDesktopClient,
@@ -551,35 +546,71 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 
 func leaseNextQueuedMonitorTaskSQL() string {
 	return `
-		WITH candidate AS (
-			SELECT dt.desktop_id
-			FROM desktop_tasks AS dt
-			WHERE dt.tenant_id = $1
+		WITH current_accounts AS (
+			SELECT desktop_id, user_id, platform_id, platform_uid, account_fingerprint
+			FROM platform_accounts
+			WHERE tenant_id = $1
+			  AND workspace_id = $2
+			  AND client_id = $3
+			  AND deleted_at IS NULL
+			  AND delete_requested_at IS NULL
+		),
+		candidate_pool AS (
+			SELECT dt.desktop_id,
+				ca.desktop_id AS lease_account_id,
+				dt.lane_weight,
+				dt.priority,
+				dt.enqueued_at,
+				dt.created_at,
+				` + monitorAccountLeaseLockSQL("target_account") + ` AS account_slot_lock_key
+			FROM current_accounts AS ca
+			JOIN platform_accounts AS target_account
+			  ON target_account.tenant_id = $1
+			 AND target_account.workspace_id = $2
+			 AND target_account.user_id = ca.user_id
+			 AND target_account.platform_id = ca.platform_id
+			 AND ` + monitorAccountIdentityMatchSQL("ca", "target_account") + `
+			JOIN desktop_tasks AS dt
+			  ON dt.target_account_id = target_account.desktop_id
+			 AND dt.tenant_id = $1
 			 AND dt.workspace_id = $2
+			 AND dt.platform_id = ca.platform_id
 			 AND dt.kind = 'monitor'
 			 AND dt.status = 'queued'
-			  AND NOT EXISTS (
+			WHERE NOT EXISTS (
 				SELECT 1
 				FROM desktop_tasks AS active
+				JOIN platform_accounts AS active_account
+				  ON active_account.desktop_id = active.target_account_id
+				 AND active_account.tenant_id = active.tenant_id
+				 AND active_account.workspace_id = active.workspace_id
 				WHERE active.tenant_id = dt.tenant_id
 				  AND active.workspace_id = dt.workspace_id
 				  AND active.kind = 'monitor'
-				  AND active.target_client_id = $3
 				  AND active.platform_id = dt.platform_id
 				  AND active.status = 'in_progress'
 				  AND active.desktop_id <> dt.desktop_id
+				  AND active_account.user_id = target_account.user_id
+				  AND ` + monitorAccountIdentityMatchSQL("active_account", "target_account") + `
 			  )
 			  AND NOT EXISTS (
 				SELECT 1
 				FROM desktop_tasks AS recent
+				JOIN platform_accounts AS recent_account
+				  ON recent_account.desktop_id = recent.target_account_id
+				 AND recent_account.tenant_id = recent.tenant_id
+				 AND recent_account.workspace_id = recent.workspace_id
 				WHERE recent.tenant_id = dt.tenant_id
 				  AND recent.workspace_id = dt.workspace_id
 				  AND recent.kind = 'monitor'
-				  AND recent.target_client_id = $3
 				  AND recent.platform_id = dt.platform_id
-				  AND recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')
-				  AND recent.updated_at >= now() - interval '30 seconds'
+				  AND (
+					(recent.status = 'succeeded' AND recent.updated_at >= now() - interval '` + monitorSucceededTaskCooldownSQL + `')
+					OR (recent.status IN ('failed', 'unknown', 'aborted') AND recent.updated_at >= now() - interval '` + monitorUnhealthyTaskCooldownSQL + `')
+				  )
 				  AND recent.desktop_id <> dt.desktop_id
+				  AND recent_account.user_id = target_account.user_id
+				  AND ` + monitorAccountIdentityMatchSQL("recent_account", "target_account") + `
 			  )
 			  AND (
 				SELECT COUNT(DISTINCT active_platform.platform_id)
@@ -590,23 +621,44 @@ func leaseNextQueuedMonitorTaskSQL() string {
 				  AND active_platform.target_client_id = $3
 				  AND active_platform.status = 'in_progress'
 				  AND active_platform.desktop_id <> dt.desktop_id
-			  ) < $8::integer
-			  AND (
-				dt.target_client_id = $3
-				OR ($4::boolean AND dt.target_account_id = ANY($5::uuid[]))
+			  ) < $6::integer
+			  AND NOT EXISTS (
+				SELECT 1
+				FROM desktop_tasks AS active_client_platform
+				WHERE active_client_platform.tenant_id = dt.tenant_id
+				  AND active_client_platform.workspace_id = dt.workspace_id
+				  AND active_client_platform.kind = 'monitor'
+				  AND active_client_platform.target_client_id = $3
+				  AND active_client_platform.platform_id = dt.platform_id
+				  AND active_client_platform.status = 'in_progress'
+				  AND active_client_platform.desktop_id <> dt.desktop_id
 			  )
 			ORDER BY dt.lane_weight DESC,
 				dt.priority DESC,
 				COALESCE(dt.enqueued_at, dt.created_at) ASC,
 				dt.created_at ASC,
-				dt.desktop_id ASC
+				dt.desktop_id ASC,
+				ca.desktop_id ASC
+			LIMIT ` + monitorLeaseCandidateScanLimitSQL + `
+			FOR UPDATE OF dt SKIP LOCKED
+		),
+		candidate AS (
+			SELECT desktop_id, lease_account_id
+			FROM candidate_pool
+			WHERE pg_try_advisory_xact_lock(account_slot_lock_key)
+			ORDER BY lane_weight DESC,
+				priority DESC,
+				COALESCE(enqueued_at, created_at) ASC,
+				created_at ASC,
+				desktop_id ASC,
+				lease_account_id ASC
 			LIMIT 1
-			FOR UPDATE SKIP LOCKED
 		)
 		UPDATE desktop_tasks AS t
 		SET target_client_id = $3,
-		    active_attempt_id = $6,
-		    lease_token_hash = $7,
+		    target_account_id = candidate.lease_account_id,
+		    active_attempt_id = $4,
+		    lease_token_hash = $5,
 		    lease_expires_at = now() + interval '10 minutes',
 		    status = 'in_progress',
 		    attempts = t.attempts + 1,
@@ -616,6 +668,51 @@ func leaseNextQueuedMonitorTaskSQL() string {
 		RETURNING ` + desktopTaskRepositoryReturningColumns
 }
 
+func monitorAccountLeaseLockSQL(alias string) string {
+	return fmt.Sprintf(`hashtextextended(
+				concat_ws(
+					':',
+					'monitor_desktop_account_slot',
+					%s.tenant_id::text,
+					%s.workspace_id::text,
+					%s.user_id::text,
+					%s.platform_id,
+					CASE
+						WHEN btrim(COALESCE(%s.account_fingerprint, '')) <> '' THEN 'fp:' || %s.account_fingerprint
+						WHEN btrim(COALESCE(%s.platform_uid, '')) <> '' THEN 'uid:' || %s.platform_uid
+						ELSE 'id:' || %s.desktop_id::text
+					END
+				),
+				0
+			)`,
+		alias,
+		alias,
+		alias,
+		alias,
+		alias, alias,
+		alias, alias,
+		alias,
+	)
+}
+
+func monitorAccountIdentityMatchSQL(leftAlias, rightAlias string) string {
+	return fmt.Sprintf(`(
+				%s.desktop_id = %s.desktop_id
+				OR (
+					btrim(COALESCE(%s.account_fingerprint, '')) <> ''
+					AND %s.account_fingerprint = %s.account_fingerprint
+				)
+				OR (
+					btrim(COALESCE(%s.platform_uid, '')) <> ''
+					AND %s.platform_uid = %s.platform_uid
+				)
+			)`,
+		leftAlias, rightAlias,
+		leftAlias, leftAlias, rightAlias,
+		leftAlias, leftAlias, rightAlias,
+	)
+}
+
 func (s *DesktopTaskService) leaseNextQueuedPublishTask(
 	ctx context.Context,
 	client *repository.DesktopClient,
@@ -679,12 +776,6 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 	desktopID uuid.UUID,
 	params repository.DesktopTaskLeaseParams,
 ) (*repository.DesktopTask, error) {
-	accountIDs, err := s.loadMonitorLeaseAccountIDs(ctx, client)
-	if err != nil {
-		return nil, err
-	}
-	hasAccountIDs := len(accountIDs) > 0
-
 	tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
 	if err != nil {
 		return nil, err
@@ -696,8 +787,6 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 		client.WorkspaceID,
 		client.TenantID,
 		client.ID,
-		hasAccountIDs,
-		accountIDs,
 		params.AttemptID,
 		params.LeaseTokenHash,
 		desktopPublishMaxAttempts,
@@ -715,13 +804,35 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 
 func leaseQueuedDesktopTaskByIDSQL() string {
 	return `
-		WITH candidate AS (
-			SELECT dt.desktop_id
+		WITH current_accounts AS (
+			SELECT desktop_id, user_id, platform_id, platform_uid, account_fingerprint
+			FROM platform_accounts
+			WHERE tenant_id = $3
+			  AND workspace_id = $2
+			  AND client_id = $4
+			  AND deleted_at IS NULL
+			  AND delete_requested_at IS NULL
+		),
+		candidate AS (
+			SELECT dt.desktop_id,
+				dt.kind,
+				ca.desktop_id AS lease_account_id,
+				` + monitorAccountLeaseLockSQL("target_account") + ` AS account_slot_lock_key
 			FROM desktop_tasks AS dt
+			LEFT JOIN platform_accounts AS target_account
+			  ON dt.kind = 'monitor'
+			 AND target_account.desktop_id = dt.target_account_id
+			 AND target_account.tenant_id = dt.tenant_id
+			 AND target_account.workspace_id = dt.workspace_id
+			LEFT JOIN current_accounts AS ca
+			  ON dt.kind = 'monitor'
+			 AND ca.user_id = target_account.user_id
+			 AND ca.platform_id = dt.platform_id
+			 AND ` + monitorAccountIdentityMatchSQL("ca", "target_account") + `
 			WHERE dt.desktop_id = $1
 			  AND dt.workspace_id = $2
 			  AND dt.status = 'queued'
-			  AND (dt.kind <> 'publish' OR dt.attempts < $9)
+			  AND (dt.kind <> 'publish' OR dt.attempts < $7)
 			  AND (
 				dt.kind <> 'publish'
 				OR EXISTS (
@@ -734,18 +845,27 @@ func leaseQueuedDesktopTaskByIDSQL() string {
 					  AND pa.delete_requested_at IS NULL
 				)
 			  )
+			  AND (
+				dt.kind <> 'monitor'
+				OR ca.desktop_id IS NOT NULL
+			  )
 			  AND (
 				dt.kind <> 'monitor'
 				OR NOT EXISTS (
 					SELECT 1
 					FROM desktop_tasks AS active
+					JOIN platform_accounts AS active_account
+					  ON active_account.desktop_id = active.target_account_id
+					 AND active_account.tenant_id = active.tenant_id
+					 AND active_account.workspace_id = active.workspace_id
 					WHERE active.tenant_id = dt.tenant_id
 					  AND active.workspace_id = dt.workspace_id
 					  AND active.kind = 'monitor'
-					  AND active.target_client_id = $4
 					  AND active.platform_id = dt.platform_id
 					  AND active.status = 'in_progress'
 					  AND active.desktop_id <> dt.desktop_id
+					  AND active_account.user_id = target_account.user_id
+					  AND ` + monitorAccountIdentityMatchSQL("active_account", "target_account") + `
 				)
 			  )
 			  AND (
@@ -753,14 +873,21 @@ func leaseQueuedDesktopTaskByIDSQL() string {
 				OR NOT EXISTS (
 					SELECT 1
 					FROM desktop_tasks AS recent
+					JOIN platform_accounts AS recent_account
+					  ON recent_account.desktop_id = recent.target_account_id
+					 AND recent_account.tenant_id = recent.tenant_id
+					 AND recent_account.workspace_id = recent.workspace_id
 					WHERE recent.tenant_id = dt.tenant_id
 					  AND recent.workspace_id = dt.workspace_id
 					  AND recent.kind = 'monitor'
-					  AND recent.target_client_id = $4
 					  AND recent.platform_id = dt.platform_id
-					  AND recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')
-					  AND recent.updated_at >= now() - interval '30 seconds'
+					  AND (
+						(recent.status = 'succeeded' AND recent.updated_at >= now() - interval '` + monitorSucceededTaskCooldownSQL + `')
+						OR (recent.status IN ('failed', 'unknown', 'aborted') AND recent.updated_at >= now() - interval '` + monitorUnhealthyTaskCooldownSQL + `')
+					  )
 					  AND recent.desktop_id <> dt.desktop_id
+					  AND recent_account.user_id = target_account.user_id
+					  AND ` + monitorAccountIdentityMatchSQL("recent_account", "target_account") + `
 				)
 			  )
 			  AND (
@@ -774,36 +901,55 @@ func leaseQueuedDesktopTaskByIDSQL() string {
 					  AND active_platform.target_client_id = $4
 					  AND active_platform.status = 'in_progress'
 					  AND active_platform.desktop_id <> dt.desktop_id
-				) < $10::integer
+				) < $8::integer
+			  )
+			  AND (
+				dt.kind <> 'monitor'
+				OR NOT EXISTS (
+					SELECT 1
+					FROM desktop_tasks AS active_client_platform
+					WHERE active_client_platform.tenant_id = dt.tenant_id
+					  AND active_client_platform.workspace_id = dt.workspace_id
+					  AND active_client_platform.kind = 'monitor'
+					  AND active_client_platform.target_client_id = $4
+					  AND active_client_platform.platform_id = dt.platform_id
+					  AND active_client_platform.status = 'in_progress'
+					  AND active_client_platform.desktop_id <> dt.desktop_id
+				)
 			  )
 			  AND (
 				(
 					dt.kind = 'monitor'
 					AND dt.tenant_id = $3
-					AND (
-						dt.target_client_id = $4
-						OR ($5::boolean AND dt.target_account_id = ANY($6::uuid[]))
-					)
 				)
 				OR (
 					dt.kind <> 'monitor'
 					AND dt.target_client_id = $4
 				)
 			  )
+			ORDER BY ca.desktop_id ASC NULLS LAST
+			LIMIT 1
+			FOR UPDATE OF dt SKIP LOCKED
+		),
+		locked_candidate AS (
+			SELECT desktop_id, lease_account_id
+			FROM candidate
+			WHERE kind <> 'monitor'
+			   OR pg_try_advisory_xact_lock(account_slot_lock_key)
 			LIMIT 1
-			FOR UPDATE SKIP LOCKED
 		)
 		UPDATE desktop_tasks AS t
 		SET target_client_id = CASE WHEN t.kind = 'monitor' THEN $4 ELSE t.target_client_id END,
-		    active_attempt_id = $7,
-		    lease_token_hash = $8,
+		    target_account_id = CASE WHEN t.kind = 'monitor' THEN locked_candidate.lease_account_id ELSE t.target_account_id END,
+		    active_attempt_id = $5,
+		    lease_token_hash = $6,
 		    lease_expires_at = now() + CASE WHEN t.kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
 		    status = 'in_progress',
 		    attempts = t.attempts + 1,
 		    started_at = COALESCE(t.started_at, now()),
 		    updated_at = now()
-		FROM candidate
-		WHERE t.desktop_id = candidate.desktop_id
+		FROM locked_candidate
+		WHERE t.desktop_id = locked_candidate.desktop_id
 		RETURNING ` + desktopTaskRepositoryReturningColumns
 }
 
@@ -822,39 +968,6 @@ func beginDesktopMonitorLeaseTx(ctx context.Context, pool *pgxpool.Pool, clientI
 	return tx, nil
 }
 
-func (s *DesktopTaskService) loadMonitorLeaseAccountIDs(ctx context.Context, client *repository.DesktopClient) ([]uuid.UUID, error) {
-	if s == nil || client == nil || s.pool == nil {
-		return nil, nil
-	}
-
-	accountIDs := loadTrackedDesktopClientAccountIDs(ctx, s.redis, client.ID)
-	rows, err := s.pool.Query(ctx, `
-		SELECT desktop_id
-		FROM platform_accounts
-		WHERE tenant_id = $1
-		  AND workspace_id = $2
-		  AND client_id = $3
-		  AND deleted_at IS NULL
-		  AND platform_id = ANY($4::text[])
-	`, client.TenantID, client.WorkspaceID, client.ID, monitoringPlatformIDs(defaultMonitoringPlatforms))
-	if err != nil {
-		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to resolve monitor lease accounts")
-	}
-	defer rows.Close()
-
-	for rows.Next() {
-		var accountID uuid.UUID
-		if scanErr := rows.Scan(&accountID); scanErr != nil {
-			return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to parse monitor lease accounts")
-		}
-		accountIDs = append(accountIDs, accountID)
-	}
-	if err := rows.Err(); err != nil {
-		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to iterate monitor lease accounts")
-	}
-	return uniqueUUIDs(accountIDs), nil
-}
-
 func scanRepositoryDesktopTask(row desktopTaskRepositoryScanner) (*repository.DesktopTask, error) {
 	var (
 		task              repository.DesktopTask
@@ -1656,16 +1769,28 @@ func (s *DesktopTaskService) clientCanLeaseMonitorTask(ctx context.Context, clie
 	if s == nil || client == nil || task == nil || task.Kind != "monitor" || task.TargetAccountID == uuid.Nil {
 		return false
 	}
-	accountIDs, err := s.loadMonitorLeaseAccountIDs(ctx, client)
-	if err != nil {
+	var canLease bool
+	if err := s.pool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1
+			FROM platform_accounts AS target_account
+			JOIN platform_accounts AS client_account
+			  ON client_account.tenant_id = target_account.tenant_id
+			 AND client_account.workspace_id = target_account.workspace_id
+			 AND client_account.user_id = target_account.user_id
+			 AND client_account.platform_id = target_account.platform_id
+			 AND client_account.client_id = $4
+			 AND client_account.deleted_at IS NULL
+			 AND client_account.delete_requested_at IS NULL
+			 AND `+monitorAccountIdentityMatchSQL("client_account", "target_account")+`
+			WHERE target_account.desktop_id = $1
+			  AND target_account.tenant_id = $2
+			  AND target_account.workspace_id = $3
+		)
+	`, task.TargetAccountID, task.TenantID, task.WorkspaceID, client.ID).Scan(&canLease); err != nil {
 		return false
 	}
-	for _, accountID := range accountIDs {
-		if accountID == task.TargetAccountID {
-			return true
-		}
-	}
-	return false
+	return canLease
 }
 
 func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {

server/internal/tenant/app/desktop_task_service_test.go

@@ -239,21 +239,33 @@ func TestBuildCountPublishTasksByStatusesQueryHidesDeletedPublishRecords(t *test
 	}
 }
 
-func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T) {
+func TestLeaseNextQueuedMonitorTaskSQLUsesAccountIdentityAndClientSlots(t *testing.T) {
 	t.Parallel()
 
 	normalized := normalizeSQLForDesktopTaskTest(leaseNextQueuedMonitorTaskSQL())
 	for _, fragment := range []string{
+		"WITH current_accounts AS",
+		"client_id = $3",
+		"dt.target_account_id = target_account.desktop_id",
+		"target_account.account_fingerprint",
+		"target_account.platform_uid",
+		"hashtextextended",
+		"pg_try_advisory_xact_lock(account_slot_lock_key)",
+		"SET target_client_id = $3, target_account_id = candidate.lease_account_id",
 		"NOT EXISTS ( SELECT 1 FROM desktop_tasks AS active",
-		"active.target_client_id = $3",
 		"active.platform_id = dt.platform_id",
 		"active.status = 'in_progress'",
 		"active.desktop_id <> dt.desktop_id",
 		"COUNT(DISTINCT active_platform.platform_id)",
 		"active_platform.target_client_id = $3",
-		"< $8::integer",
-		"recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')",
+		"< $6::integer",
+		"active_client_platform.target_client_id = $3",
+		"active_client_platform.platform_id = dt.platform_id",
+		"recent.status = 'succeeded'",
+		"recent.updated_at >= now() - interval '2 seconds'",
+		"recent.status IN ('failed', 'unknown', 'aborted')",
 		"recent.updated_at >= now() - interval '30 seconds'",
+		"ORDER BY dt.lane_weight DESC, dt.priority DESC, COALESCE(dt.enqueued_at, dt.created_at) ASC",
 	} {
 		if !strings.Contains(normalized, fragment) {
 			t.Fatalf("monitor lease query missing %q: %s", fragment, normalized)
@@ -261,20 +273,33 @@ func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T)
 	}
 }
 
-func TestLeaseQueuedDesktopTaskByIDSQLSerializesMonitorPerClientPlatform(t *testing.T) {
+func TestLeaseQueuedDesktopTaskByIDSQLUsesAccountIdentityAndClientSlots(t *testing.T) {
 	t.Parallel()
 
 	normalized := normalizeSQLForDesktopTaskTest(leaseQueuedDesktopTaskByIDSQL())
 	for _, fragment := range []string{
+		"WITH current_accounts AS",
+		"client_id = $4",
+		"dt.kind <> 'monitor' OR ca.desktop_id IS NOT NULL",
+		"target_account.account_fingerprint",
+		"target_account.platform_uid",
+		"SELECT dt.desktop_id, dt.kind, ca.desktop_id AS lease_account_id",
+		"locked_candidate AS",
+		"pg_try_advisory_xact_lock(account_slot_lock_key)",
+		"FROM locked_candidate WHERE t.desktop_id = locked_candidate.desktop_id",
+		"target_account_id = CASE WHEN t.kind = 'monitor' THEN locked_candidate.lease_account_id ELSE t.target_account_id END",
 		"dt.kind <> 'monitor' OR NOT EXISTS",
-		"active.target_client_id = $4",
 		"active.platform_id = dt.platform_id",
 		"active.status = 'in_progress'",
 		"active.desktop_id <> dt.desktop_id",
 		"COUNT(DISTINCT active_platform.platform_id)",
 		"active_platform.target_client_id = $4",
-		"< $10::integer",
-		"recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')",
+		"< $8::integer",
+		"active_client_platform.target_client_id = $4",
+		"active_client_platform.platform_id = dt.platform_id",
+		"recent.status = 'succeeded'",
+		"recent.updated_at >= now() - interval '2 seconds'",
+		"recent.status IN ('failed', 'unknown', 'aborted')",
 		"recent.updated_at >= now() - interval '30 seconds'",
 	} {
 		if !strings.Contains(normalized, fragment) {

server/internal/tenant/app/monitoring_collect_outbox_worker.go

@@ -871,7 +871,6 @@ func (s *MonitoringService) enqueueCollectNowOutbox(
 	targetClientIDs []uuid.UUID,
 	affectedTaskCount int64,
 	interruptGeneration int,
-	interruptTaskIDs []int64,
 ) error {
 	if tx == nil {
 		return nil
@@ -896,28 +895,6 @@ func (s *MonitoringService) enqueueCollectNowOutbox(
 		}
 	}
 
-	seenTaskIDs := make(map[int64]struct{}, len(interruptTaskIDs))
-	for _, taskID := range interruptTaskIDs {
-		if taskID <= 0 {
-			continue
-		}
-		if _, exists := seenTaskIDs[taskID]; exists {
-			continue
-		}
-		seenTaskIDs[taskID] = struct{}{}
-		for _, targetClientID := range targetClientIDs {
-			if _, err := tx.Exec(ctx, `
-				INSERT INTO monitoring_collect_dispatch_outbox (
-					request_id, workspace_id, target_client_id, event_kind, task_id,
-					lane, priority, interrupt_generation, reason
-				)
-				VALUES ($1, $2, $3, 'interrupt_requested', $4, 'high', 5000, $5, 'collect_now_preempt')
-			`, requestID, workspaceID, targetClientID, taskID, interruptGeneration); err != nil {
-				return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now interrupt signal")
-			}
-		}
-	}
-
 	return nil
 }
 

server/internal/tenant/app/monitoring_daily_task_worker_test.go

@@ -504,6 +504,8 @@ type captureDailyMonitorTaskTx struct {
 	execCalled bool
 	execSQL    string
 	execArgs   []any
+	execSQLs   []string
+	execArgses [][]any
 	commandTag pgconn.CommandTag
 }
 
@@ -524,6 +526,8 @@ func (tx *captureDailyMonitorTaskTx) Exec(_ context.Context, sql string, argumen
 	tx.execCalled = true
 	tx.execSQL = sql
 	tx.execArgs = arguments
+	tx.execSQLs = append(tx.execSQLs, sql)
+	tx.execArgses = append(tx.execArgses, arguments)
 	return tx.commandTag, nil
 }
 func (tx *captureDailyMonitorTaskTx) Query(context.Context, string, ...any) (pgx.Rows, error) {

server/internal/tenant/app/monitoring_phase2_desktop_tasks.go

@@ -12,7 +12,6 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	"go.uber.org/zap"
 
-	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
 	"github.com/geo-platform/tenant-api/internal/shared/response"
 	"github.com/geo-platform/tenant-api/internal/shared/stream"
 	"github.com/geo-platform/tenant-api/internal/tenant/repository"
@@ -66,14 +65,6 @@ type monitorDesktopTaskTargetCandidate struct {
 	ClientOrder          int
 }
 
-type monitorDesktopInterruptTarget struct {
-	TaskID              string
-	MonitorTaskID       int64
-	TargetClientID      string
-	WorkspaceID         int64
-	InterruptGeneration int
-}
-
 func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 {
 	if len(specs) == 0 {
 		return nil
@@ -747,84 +738,6 @@ func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Cont
 	return nil
 }
 
-func (s *MonitoringService) requestPhase2MonitorInterrupts(
-	ctx context.Context,
-	targetClientID uuid.UUID,
-	excludedMonitorTaskIDs []int64,
-	excludedPlatformIDs []string,
-	interruptGeneration int,
-) ([]monitorDesktopInterruptTarget, error) {
-	if s == nil || s.businessPool == nil {
-		return nil, nil
-	}
-
-	queryArgs := []any{targetClientID, interruptGeneration}
-	nextParam := 3
-	query := `
-		UPDATE desktop_tasks
-		SET control_flags = (
-				CASE
-					WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object'
-						THEN '{}'::jsonb
-					ELSE control_flags
-				END
-			) || jsonb_build_object(
-				'interrupt_requested', true,
-				'interrupt_generation', $2::integer,
-				'interrupt_reason', 'collect_now_preempt'
-			),
-		    interrupt_generation = GREATEST(interrupt_generation, $2::integer),
-		    interrupted_at = COALESCE(interrupted_at, NOW()),
-		    interrupt_reason = 'collect_now_preempt',
-		    updated_at = NOW()
-		WHERE kind = 'monitor'
-		  AND target_client_id = $1
-		  AND status = 'in_progress'
-		  AND lane IN ('normal', 'normal_boosted', 'retry')
-		  AND monitor_task_id IS NOT NULL
-	`
-	if len(excludedMonitorTaskIDs) > 0 {
-		query += fmt.Sprintf(` AND NOT (monitor_task_id = ANY($%d::bigint[]))`, nextParam)
-		queryArgs = append(queryArgs, excludedMonitorTaskIDs)
-		nextParam++
-	}
-	excludedPlatformIDs = reconcileEnabledMonitoringPlatforms(excludedPlatformIDs)
-	if len(excludedPlatformIDs) > 0 {
-		query += fmt.Sprintf(` AND NOT (platform_id = ANY($%d::text[]))`, nextParam)
-		queryArgs = append(queryArgs, excludedPlatformIDs)
-	}
-	query += `
-		RETURNING desktop_id::text, monitor_task_id, target_client_id::text, workspace_id, interrupt_generation
-	`
-
-	rows, err := s.businessPool.Query(ctx, query, queryArgs...)
-	if err != nil {
-		if s.logger != nil {
-			s.logger.Error("phase2 monitor interrupt query failed",
-				zap.Error(err),
-				zap.String("target_client_id", targetClientID.String()),
-				zap.Int("interrupt_generation", interruptGeneration),
-				zap.Int("excluded_monitor_task_count", len(excludedMonitorTaskIDs)),
-			)
-		}
-		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to mark active phase2 monitor desktop tasks for interrupt")
-	}
-	defer rows.Close()
-
-	result := make([]monitorDesktopInterruptTarget, 0)
-	for rows.Next() {
-		var item monitorDesktopInterruptTarget
-		if scanErr := rows.Scan(&item.TaskID, &item.MonitorTaskID, &item.TargetClientID, &item.WorkspaceID, &item.InterruptGeneration); scanErr != nil {
-			return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to parse active phase2 monitor interrupt targets")
-		}
-		result = append(result, item)
-	}
-	if err := rows.Err(); err != nil {
-		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to iterate active phase2 monitor interrupt targets")
-	}
-	return result, nil
-}
-
 func (s *MonitoringService) deferQueuedPhase2MonitorTasks(
 	ctx context.Context,
 	targetClientID uuid.UUID,
@@ -953,36 +866,6 @@ func publishMonitorDesktopTaskAvailable(
 	return nil
 }
 
-func publishPhase2MonitorControlEvent(
-	ctx context.Context,
-	client *rabbitmq.Client,
-	logger *zap.Logger,
-	target monitorDesktopInterruptTarget,
-) {
-	if client == nil || strings.TrimSpace(target.TargetClientID) == "" || strings.TrimSpace(target.TaskID) == "" {
-		return
-	}
-	if err := publishDesktopDispatchEvent(ctx, client, logger, stream.DesktopDispatchEvent{
-		Type:                "task_control",
-		TaskID:              target.TaskID,
-		WorkspaceID:         target.WorkspaceID,
-		TargetClientID:      target.TargetClientID,
-		Status:              "in_progress",
-		Kind:                "monitor",
-		Lane:                "high",
-		InterruptGeneration: target.InterruptGeneration,
-		Control:             "interrupt_requested",
-		Reason:              "collect_now_preempt",
-		UpdatedAt:           time.Now().UTC(),
-	}); err != nil && logger != nil {
-		logger.Warn("phase2 monitor control event publish failed",
-			zap.String("task_id", target.TaskID),
-			zap.Int64("monitor_task_id", target.MonitorTaskID),
-			zap.Error(err),
-		)
-	}
-}
-
 func monitoringLaneWeight(lane string) int {
 	switch strings.TrimSpace(lane) {
 	case "high":

server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go

@@ -206,17 +206,3 @@ func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) {
 	assert.Equal(t, "skipped", outcome.TaskStatus)
 	assert.Equal(t, "stale_business_date", outcome.StoredSkipReason)
 }
-
-func TestRequestPhase2MonitorInterruptsAcceptsExcludedPlatforms(t *testing.T) {
-	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701")
-	targets, err := (&MonitoringService{}).requestPhase2MonitorInterrupts(
-		t.Context(),
-		clientID,
-		[]int64{101, 102},
-		[]string{"doubao", "qwen"},
-		7,
-	)
-
-	require.NoError(t, err)
-	assert.Empty(t, targets)
-}

server/internal/tenant/app/monitoring_service.go

@@ -882,7 +882,7 @@ func (s *MonitoringService) CollectNow(
 		return nil, err
 	}
 
-	refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, err := s.ensureCollectNowTasks(
+	refreshedCount, createdCount, leasedCount, completedCount, err := s.ensureCollectNowTasks(
 		ctx,
 		tx,
 		actor.TenantID,
@@ -969,7 +969,7 @@ func (s *MonitoringService) CollectNow(
 			$12, $13, $14,
 			$15, $16
 		)
-	`, requestID, actor.TenantID, workspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(len(interruptTaskIDs)), interruptGeneration, message, ttlExpiresAt); err != nil {
+	`, requestID, actor.TenantID, workspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(0), interruptGeneration, message, ttlExpiresAt); err != nil {
 		return nil, response.ErrInternal(50041, "request_create_failed", "failed to persist monitoring collect-now request")
 	}
 
@@ -981,12 +981,6 @@ func (s *MonitoringService) CollectNow(
 		dispatchTargetClientIDs,
 		affectedTaskCount,
 		interruptGeneration,
-		func() []int64 {
-			if !options.Preempt {
-				return nil
-			}
-			return collectNowInterruptTaskIDs(interruptTaskIDs)
-		}(),
 	); err != nil {
 		return nil, err
 	}
@@ -1022,14 +1016,11 @@ func (s *MonitoringService) CollectNow(
 		}
 	}
 
-	phase2InterruptTargets := make([]monitorDesktopInterruptTarget, 0)
 	phase2DeferredTasks := make([]*repository.DesktopTask, 0)
 	if options.Preempt && phase2DispatchReady {
 		excludedMonitorTaskIDs := make([]int64, 0, len(phase2TaskSpecs))
-		excludedPlatformIDs := make([]string, 0, len(phase2TaskSpecs))
 		for _, spec := range phase2TaskSpecs {
 			excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID)
-			excludedPlatformIDs = append(excludedPlatformIDs, spec.PlatformID)
 		}
 		phase2TargetClientIDs := desktopTaskClientIDs(phase2PublishedTasks)
 		if len(phase2TargetClientIDs) == 0 {
@@ -1041,28 +1032,9 @@ func (s *MonitoringService) CollectNow(
 				return nil, deferErr
 			}
 			phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...)
-			interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, excludedPlatformIDs, interruptGeneration)
-			if interruptErr != nil {
-				return nil, interruptErr
-			}
-			phase2InterruptTargets = append(phase2InterruptTargets, interruptTargets...)
 		}
 		if len(phase2DeferredTasks) > 0 {
 			abortedQueuedCount += int64(len(phase2DeferredTasks))
-		}
-		if len(phase2InterruptTargets) > 0 {
-			// Keep collect-now request counters in sync after per-platform dispatch
-			// fan-out chooses more than one desktop client.
-			if _, updateErr := s.monitoringPool.Exec(ctx, `
-				UPDATE monitoring_collect_requests
-				SET aborted_queued_count = aborted_queued_count + $2,
-				    interrupt_requested_count = interrupt_requested_count + $3,
-				    updated_at = NOW()
-				WHERE request_id = $1
-			`, requestID, len(phase2DeferredTasks), len(phase2InterruptTargets)); updateErr != nil {
-				return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 preempt counts")
-			}
-		} else if len(phase2DeferredTasks) > 0 {
 			if _, updateErr := s.monitoringPool.Exec(ctx, `
 				UPDATE monitoring_collect_requests
 				SET aborted_queued_count = aborted_queued_count + $2,
@@ -1107,12 +1079,8 @@ func (s *MonitoringService) CollectNow(
 			)
 		}
 	}
-	for _, target := range phase2InterruptTargets {
-		publishPhase2MonitorControlEvent(context.Background(), s.rabbitMQ, s.logger, target)
-	}
 
 	hasEffectiveSnapshot := leasedCount > 0 || completedCount > 0
-	totalInterruptRequestedCount := int64(len(interruptTaskIDs) + len(phase2InterruptTargets))
 	responseData := &MonitoringCollectNowResponse{
 		CollectionMode:          quota.CollectionMode,
 		RefreshedTaskCount:      refreshedCount,
@@ -1125,7 +1093,7 @@ func (s *MonitoringService) CollectNow(
 		AffectedTaskCount:       affectedTaskCount,
 		PromotedTaskCount:       affectedTaskCount,
 		AbortedQueuedCount:      abortedQueuedCount,
-		InterruptRequestedCount: totalInterruptRequestedCount,
+		InterruptRequestedCount: 0,
 		InterruptGeneration:     interruptGeneration,
 		TTLExpiresAt:            ttlExpiresAt.Format(time.RFC3339),
 		Message:                 message,
@@ -1698,7 +1666,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
 	targetClientID uuid.UUID,
 	interruptGeneration int,
 	executionOwner string,
-) (int64, int64, int64, int64, []int64, error) {
+) (int64, int64, int64, int64, error) {
 	const runMode = "plugin_standard"
 	if strings.TrimSpace(executionOwner) == "" {
 		executionOwner = "legacy"
@@ -1707,12 +1675,11 @@ func (s *MonitoringService) ensureCollectNowTasks(
 	dateText := businessDate.Format("2006-01-02")
 	var refreshedCount int64
 	var createdCount int64
-	interruptTaskIDs := make([]int64, 0)
 
 	for _, question := range questions {
 		questionText, textErr := monitoringQuestionTextSnapshot(question)
 		if textErr != nil {
-			return 0, 0, 0, 0, nil, textErr
+			return 0, 0, 0, 0, textErr
 		}
 		for _, platform := range platforms {
 			platformID := normalizeMonitoringPlatformID(platform.ID)
@@ -1762,11 +1729,11 @@ func (s *MonitoringService) ensureCollectNowTasks(
 				  AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped')
 			`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner, questionText)
 			if err != nil {
-				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
+				return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
 			}
 			refreshedCount += tag.RowsAffected()
 
-			rows, err := tx.Query(ctx, `
+			tag, err = tx.Exec(ctx, `
 				UPDATE monitoring_collect_tasks
 				SET question_hash = $7,
 				    question_text_snapshot = $11,
@@ -1793,24 +1760,10 @@ func (s *MonitoringService) ensureCollectNowTasks(
 				  AND business_date = $8::date
 				  AND status = 'leased'
 				  AND COALESCE(execution_owner, 'legacy') = 'legacy'
-				RETURNING id
 			`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, questionText)
 			if err != nil {
-				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks")
+				return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks")
 			}
-			for rows.Next() {
-				var taskID int64
-				if scanErr := rows.Scan(&taskID); scanErr != nil {
-					rows.Close()
-					return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to parse promoted leased monitoring task")
-				}
-				interruptTaskIDs = append(interruptTaskIDs, taskID)
-			}
-			if err := rows.Err(); err != nil {
-				rows.Close()
-				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to iterate promoted leased monitoring tasks")
-			}
-			rows.Close()
 
 			tag, err = tx.Exec(ctx, `
 				INSERT INTO monitoring_collect_tasks (
@@ -1831,7 +1784,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
 				DO NOTHING
 			`, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner, questionText)
 			if err != nil {
-				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks")
+				return 0, 0, 0, 0, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks")
 			}
 			createdCount += tag.RowsAffected()
 		}
@@ -1850,10 +1803,10 @@ func (s *MonitoringService) ensureCollectNowTasks(
 	  AND business_date = $4::date
 	  AND question_id = ANY($5)
 	`, tenantID, brandID, monitoringCollectorType, dateText, configuredQuestionIDs(questions)).Scan(&leasedCount, &completedCount); err != nil {
-		return 0, 0, 0, 0, nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks")
+		return 0, 0, 0, 0, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks")
 	}
 
-	return refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, nil
+	return refreshedCount, createdCount, leasedCount, completedCount, nil
 }
 
 func (s *MonitoringService) deferQueuedNormalTasks(
@@ -2255,25 +2208,6 @@ func uuidStrings(values []uuid.UUID) []string {
 	return result
 }
 
-func collectNowInterruptTaskIDs(taskIDs []int64) []int64 {
-	if len(taskIDs) == 0 {
-		return nil
-	}
-	seen := make(map[int64]struct{}, len(taskIDs))
-	result := make([]int64, 0, len(taskIDs))
-	for _, taskID := range taskIDs {
-		if taskID <= 0 {
-			continue
-		}
-		if _, ok := seen[taskID]; ok {
-			continue
-		}
-		seen[taskID] = struct{}{}
-		result = append(result, taskID)
-	}
-	return result
-}
-
 func desktopTaskClientIDs(tasks []*repository.DesktopTask) []uuid.UUID {
 	if len(tasks) == 0 {
 		return nil

server/internal/tenant/app/monitoring_service_date_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -153,6 +154,29 @@ func TestCollectNowDispatchTargetClientIDsDeduplicatesByPlatformOrder(t *testing
 	assert.Equal(t, []uuid.UUID{sharedClientID, qwenClientID}, targets)
 }
 
+func TestEnqueueCollectNowOutboxDoesNotInterruptActiveTasks(t *testing.T) {
+	ctx := t.Context()
+	targetClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000503")
+	tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("INSERT 0 1")}
+
+	err := (&MonitoringService{}).enqueueCollectNowOutbox(
+		ctx,
+		tx,
+		uuid.New(),
+		11,
+		[]uuid.UUID{targetClientID},
+		3,
+		9,
+	)
+
+	require.NoError(t, err)
+	require.Len(t, tx.execSQLs, 1)
+	normalized := normalizeSQLWhitespace(tx.execSQLs[0])
+	assert.Contains(t, normalized, "'dispatch_high'")
+	assert.NotContains(t, normalized, "'interrupt_requested'")
+	assert.NotContains(t, normalized, "task_control")
+}
+
 func TestMonitoringPlatformAuthorizationStatus(t *testing.T) {
 	clientID := uuid.New()
 

server/migrations/20260622190000_add_monitor_desktop_lease_indexes.down.sql

@@ -0,0 +1,5 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_platform_accounts_monitor_identity_uid;
+DROP INDEX CONCURRENTLY IF EXISTS idx_platform_accounts_monitor_identity_fingerprint;
+DROP INDEX CONCURRENTLY IF EXISTS idx_desktop_tasks_monitor_active_client_platform;
+DROP INDEX CONCURRENTLY IF EXISTS idx_desktop_tasks_monitor_active_platform_account;
+DROP INDEX CONCURRENTLY IF EXISTS idx_desktop_tasks_monitor_queue_account_order;

server/migrations/20260622190000_add_monitor_desktop_lease_indexes.up.sql

@@ -0,0 +1,62 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_desktop_tasks_monitor_queue_account_order
+    ON desktop_tasks (
+        tenant_id,
+        workspace_id,
+        target_account_id,
+        platform_id,
+        lane_weight DESC,
+        priority DESC,
+        enqueued_at ASC,
+        created_at ASC,
+        desktop_id ASC
+    )
+    WHERE kind = 'monitor'
+      AND status = 'queued';
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_desktop_tasks_monitor_active_platform_account
+    ON desktop_tasks (
+        tenant_id,
+        workspace_id,
+        platform_id,
+        status,
+        target_account_id,
+        updated_at DESC
+    )
+    WHERE kind = 'monitor'
+      AND status IN ('in_progress', 'succeeded', 'failed', 'unknown', 'aborted');
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_desktop_tasks_monitor_active_client_platform
+    ON desktop_tasks (
+        tenant_id,
+        workspace_id,
+        target_client_id,
+        platform_id,
+        status,
+        desktop_id
+    )
+    WHERE kind = 'monitor'
+      AND status = 'in_progress';
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_platform_accounts_monitor_identity_fingerprint
+    ON platform_accounts (
+        tenant_id,
+        workspace_id,
+        user_id,
+        platform_id,
+        account_fingerprint
+    )
+    WHERE deleted_at IS NULL
+      AND delete_requested_at IS NULL
+      AND btrim(COALESCE(account_fingerprint, '')) <> '';
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_platform_accounts_monitor_identity_uid
+    ON platform_accounts (
+        tenant_id,
+        workspace_id,
+        user_id,
+        platform_id,
+        platform_uid
+    )
+    WHERE deleted_at IS NULL
+      AND delete_requested_at IS NULL
+      AND btrim(COALESCE(platform_uid, '')) <> '';