feat(monitoring): dispatch monitoring tasks to desktop via AMQP outbox

Replace SSE /desktop/events with priority AMQP dispatch for monitoring runs, and add phase1/phase2 infrastructure so desktop workers can lease, resume, report, skip, and cancel monitoring tasks over the existing dispatch WebSocket. - Add monitoring collect outbox worker and phase2 desktop task fields - Add /desktop/monitoring/tasks/{lease,resume,result,skip,cancel} routes - Introduce execution-devtools and network-observer on desktop runtime - Refactor runtime-controller, LoginView, and doubao adapter for the new flow - Add channelName column to tracking views Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-22 00:24:21 +08:00
parent 749b6b99cd
commit 4142c53fa6
57 changed files with 7897 additions and 985 deletions
@@ -13,9 +13,14 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"go.uber.org/zap"

 	"github.com/geo-platform/tenant-api/internal/shared/auth"
+	"github.com/geo-platform/tenant-api/internal/shared/config"
+	"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"
 )

 const (
@@ -27,14 +32,25 @@ const (
 )

 type MonitoringService struct {
-	businessPool   *pgxpool.Pool
-	monitoringPool *pgxpool.Pool
+	businessPool                  *pgxpool.Pool
+	monitoringPool                *pgxpool.Pool
+	rabbitMQ                      *rabbitmq.Client
+	logger                        *zap.Logger
+	desktopTasksRolloutPercentage int
 }

-func NewMonitoringService(businessPool, monitoringPool *pgxpool.Pool) *MonitoringService {
+func NewMonitoringService(
+	businessPool, monitoringPool *pgxpool.Pool,
+	rabbitMQClient *rabbitmq.Client,
+	dispatchCfg config.MonitoringDispatchConfig,
+	logger *zap.Logger,
+) *MonitoringService {
 	return &MonitoringService{
-		businessPool:   businessPool,
-		monitoringPool: monitoringPool,
+		businessPool:                  businessPool,
+		monitoringPool:                monitoringPool,
+		rabbitMQ:                      rabbitMQClient,
+		logger:                        logger,
+		desktopTasksRolloutPercentage: dispatchCfg.DesktopTasksRolloutPercentage,
 	}
 }

@@ -185,13 +201,28 @@ type MonitoringQuestionContentCitation struct {
 }

 type MonitoringCollectNowResponse struct {
-	CollectionMode       string `json:"collection_mode"`
-	RefreshedTaskCount   int64  `json:"refreshed_task_count"`
-	CreatedTaskCount     int64  `json:"created_task_count"`
-	LeasedTaskCount      int64  `json:"leased_task_count"`
-	CompletedTaskCount   int64  `json:"completed_task_count"`
-	HasEffectiveSnapshot bool   `json:"has_effective_snapshot"`
-	Message              string `json:"message"`
+	CollectionMode          string `json:"collection_mode"`
+	RefreshedTaskCount      int64  `json:"refreshed_task_count"`
+	CreatedTaskCount        int64  `json:"created_task_count"`
+	LeasedTaskCount         int64  `json:"leased_task_count"`
+	CompletedTaskCount      int64  `json:"completed_task_count"`
+	HasEffectiveSnapshot    bool   `json:"has_effective_snapshot"`
+	RequestID               string `json:"request_id,omitempty"`
+	TargetClientID          string `json:"target_client_id,omitempty"`
+	AffectedTaskCount       int64  `json:"affected_task_count,omitempty"`
+	PromotedTaskCount       int64  `json:"promoted_task_count,omitempty"`
+	AbortedQueuedCount      int64  `json:"aborted_queued_count,omitempty"`
+	InterruptRequestedCount int64  `json:"interrupt_requested_count,omitempty"`
+	InterruptGeneration     int    `json:"interrupt_generation,omitempty"`
+	TTLExpiresAt            string `json:"ttl_expires_at,omitempty"`
+	Message                 string `json:"message"`
+}
+
+type MonitoringCollectNowOptions struct {
+	PlatformIDs          []string
+	Preempt              bool
+	WaitForFirstDispatch bool
+	TargetClientID       *uuid.UUID
 }

 type monitoringQuotaConfig struct {
@@ -480,7 +511,12 @@ func (s *MonitoringService) QuestionDetail(ctx context.Context, brandID, questio
 	}, nil
 }

-func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywordID *int64) (*MonitoringCollectNowResponse, error) {
+func (s *MonitoringService) CollectNow(
+	ctx context.Context,
+	brandID int64,
+	keywordID *int64,
+	options MonitoringCollectNowOptions,
+) (*MonitoringCollectNowResponse, error) {
 	actor := auth.MustActor(ctx)

 	brand, err := s.resolveBrand(ctx, actor.TenantID, brandID)
@@ -522,9 +558,18 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo
 	}

 	platforms := resolveMonitoringPlatforms(quota.EnabledPlatforms)
+	platforms, err = intersectMonitoringPlatforms(platforms, options.PlatformIDs)
+	if err != nil {
+		return nil, err
+	}
 	questionIDs := configuredQuestionIDs(configuredQuestions)
+	platformIDs := monitoringPlatformIDs(platforms)
+	executionOwner := "legacy"
+	if s.shouldUseDesktopTasksExecution(actor.TenantID) {
+		executionOwner = "desktop_tasks"
+	}

-	clientID, err := s.resolveCollectNowClientID(ctx, actor.TenantID, actor.PrimaryWorkspaceID, actor.UserID)
+	clientID, err := s.resolveCollectNowClientID(ctx, actor.TenantID, actor.PrimaryWorkspaceID, actor.UserID, options.TargetClientID)
 	if err != nil {
 		return nil, err
 	}
@@ -537,7 +582,10 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo
 	}

 	now := time.Now().UTC()
-	todayTime, today := monitoringBusinessDayAndDateAt(now)
+	todayTime, _ := monitoringBusinessDayAndDateAt(now)
+	requestID := uuid.New()
+	ttlExpiresAt := now.Add(monitoringCollectNowTTL)
+	scopeHash := hashMonitoringCollectNowScope(actor.UserID, brand.ID, keywordID, questionIDs, monitoringPlatformIDs(platforms))

 	tx, err := s.monitoringPool.Begin(ctx)
 	if err != nil {
@@ -545,6 +593,58 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo
 	}
 	defer tx.Rollback(ctx)

+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, collectNowScopeLockKey(actor.UserID, brand.ID, scopeHash)); err != nil {
+		return nil, response.ErrInternal(50041, "lock_failed", "failed to acquire collect-now scope lock")
+	}
+
+	existingRequest, err := s.findActiveCollectNowRequest(
+		ctx,
+		tx,
+		actor.TenantID,
+		actor.PrimaryWorkspaceID,
+		brand.ID,
+		actor.UserID,
+		scopeHash,
+		now,
+	)
+	if err != nil {
+		return nil, err
+	}
+	if existingRequest != nil {
+		if strings.TrimSpace(existingRequest.TargetClientID) == clientID.String() {
+			base := buildCollectNowResponseFromState(quota.CollectionMode, existingRequest)
+			executing, execErr := s.hasExecutingCollectNowTask(ctx, existingRequest.RequestID)
+			if execErr != nil && s.logger != nil {
+				s.logger.Warn("collect-now executing-state inspection failed",
+					zap.String("request_id", existingRequest.RequestID.String()),
+					zap.Error(execErr),
+				)
+			}
+			if executing {
+				if base == nil {
+					base = &MonitoringCollectNowResponse{CollectionMode: quota.CollectionMode}
+				}
+				base.Message = "任务正在执行中，请等待"
+				_ = tx.Rollback(ctx)
+				return base, nil
+			}
+		}
+
+		if _, err := tx.Exec(ctx, `
+			UPDATE monitoring_collect_requests
+			SET superseded_by_request_id = $2,
+			    status = CASE
+			        WHEN status IN ('accepted', 'dispatching') THEN 'superseded'
+			        ELSE status
+			    END,
+			    updated_at = NOW()
+			WHERE request_id = $1
+			  AND superseded_by_request_id IS NULL
+		`, existingRequest.RequestID, requestID); err != nil {
+			return nil, response.ErrInternal(50041, "request_update_failed", "failed to supersede active collect-now request after client drift")
+		}
+	}
+
 	if _, err := expireMonitoringLeasedTasks(ctx, tx, &actor.TenantID, now); err != nil {
 		return nil, err
 	}
@@ -553,65 +653,255 @@ func (s *MonitoringService) CollectNow(ctx context.Context, brandID int64, keywo
 		return nil, err
 	}

-	refreshedCount, createdCount, err := s.ensureCollectNowTasks(ctx, tx, actor.TenantID, brand.ID, configuredQuestions, platforms, todayTime)
+	interruptGeneration, err := s.nextCollectNowInterruptGeneration(ctx, tx, *clientID, executionOwner)
 	if err != nil {
 		return nil, err
 	}

-	var leasedCount int64
-	var completedCount int64
-	if err := tx.QueryRow(ctx, `
-		SELECT
-			COUNT(*) FILTER (WHERE status = 'leased') AS leased_count,
-			COUNT(*) FILTER (WHERE status = 'completed') AS completed_count
-		FROM monitoring_collect_tasks
-		WHERE tenant_id = $1
-		  AND brand_id = $2
-		  AND collector_type = $3
-		  AND business_date = $4::date
-		  AND question_id = ANY($5)
-	`, actor.TenantID, brand.ID, monitoringCollectorType, today, questionIDs).Scan(&leasedCount, &completedCount); err != nil {
-		return nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks")
+	refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, err := s.ensureCollectNowTasks(
+		ctx,
+		tx,
+		actor.TenantID,
+		brand.ID,
+		configuredQuestions,
+		platforms,
+		todayTime,
+		*clientID,
+		interruptGeneration,
+		executionOwner,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	phase2TaskSpecs := make([]monitorDesktopTaskSpec, 0)
+	if executionOwner == "desktop_tasks" {
+		phase2TaskSpecs, err = s.loadMonitorDesktopTaskSpecs(
+			ctx,
+			tx,
+			actor.TenantID,
+			actor.PrimaryWorkspaceID,
+			todayTime,
+			questionIDs,
+			platformIDs,
+			*clientID,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+	abortedQueuedCount := int64(0)
+	if options.Preempt {
+		abortedQueuedCount, err = s.deferQueuedNormalTasks(
+			ctx,
+			tx,
+			actor.TenantID,
+			todayTime,
+			brand.ID,
+			configuredQuestionIDs(configuredQuestions),
+			monitoringPlatformIDs(platforms),
+			*clientID,
+			requestID,
+			ttlExpiresAt,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	requestScopeJSON, err := json.Marshal(map[string]any{
+		"brand_id":     brand.ID,
+		"keyword_id":   keywordID,
+		"question_ids": questionIDs,
+		"platform_ids": platformIDs,
+		"preempt":      options.Preempt,
+	})
+	if err != nil {
+		return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode collect-now request scope")
+	}
+	affectedTaskCount := refreshedCount + createdCount + leasedCount
+	message := "已提升为高优先级并通知桌面端优先执行"
+	if limitApplied && affectedTaskCount > 0 {
+		message = fmt.Sprintf("%s（本次最多执行 %d 个问题）", message, monitoringCollectNowQuestionLimit)
+	}
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO monitoring_collect_requests (
+			request_id, tenant_id, workspace_id, brand_id, keyword_id,
+			requested_by_user_id, target_client_id, scope_hash, request_scope,
+			status, requested_task_count, dispatched_task_count, promoted_task_count,
+			aborted_queued_count, interrupt_requested_count, interrupt_generation,
+			message, ttl_expires_at
+		)
+		VALUES (
+			$1, $2, $3, $4, $5,
+			$6, $7, $8, $9,
+			'accepted', $10, 0, $11,
+			$12, $13, $14,
+			$15, $16
+		)
+	`, requestID, actor.TenantID, actor.PrimaryWorkspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(len(interruptTaskIDs)), interruptGeneration, message, ttlExpiresAt); err != nil {
+		return nil, response.ErrInternal(50041, "request_create_failed", "failed to persist monitoring collect-now request")
+	}
+
+	if err := s.enqueueCollectNowOutbox(
+		ctx,
+		tx,
+		requestID,
+		actor.PrimaryWorkspaceID,
+		*clientID,
+		affectedTaskCount,
+		interruptGeneration,
+		func() []int64 {
+			if !options.Preempt {
+				return nil
+			}
+			return interruptTaskIDs
+		}(),
+	); err != nil {
+		return nil, err
 	}

 	if err := tx.Commit(ctx); err != nil {
 		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring collect-now")
 	}

-	hasEffectiveSnapshot := false
-	message := "已触发所选关键词立即采集，正在后台执行"
-	switch {
-	case createdCount > 0 && refreshedCount > 0:
-		message = "已补种并重置所选关键词今日任务，正在后台执行"
-	case createdCount > 0:
-		message = "已补种所选关键词的采集任务，正在后台执行"
-	case refreshedCount > 0:
-		message = "已重置所选关键词今日任务，正在后台执行"
-	case leasedCount > 0:
-		hasEffectiveSnapshot = true
-		message = "所选关键词已有执行中的任务，已继续在后台执行"
-	case completedCount > 0:
-		message = "已触发所选关键词重新采集，新的结果会覆盖今日结果"
-	default:
-		hasEffectiveSnapshot = true
-		message = "所选关键词暂无可重新调度的任务"
-	}
-	if limitApplied && (createdCount > 0 || refreshedCount > 0) {
-		message = fmt.Sprintf("%s（本次最多执行 %d 个问题）", message, monitoringCollectNowQuestionLimit)
+	_ = s.persistPrimaryClientID(ctx, actor.TenantID, actor.PrimaryWorkspaceID, *clientID)
+
+	phase2PublishedTasks := make([]*repository.DesktopTask, 0)
+	phase2DispatchReady := executionOwner == "desktop_tasks"
+	if len(phase2TaskSpecs) > 0 {
+		phase2TaskCount := len(phase2TaskSpecs)
+		publishedTasks, fallbackLegacyTaskIDs, dispatchErr := s.dispatchMonitorDesktopTasks(ctx, phase2TaskSpecs)
+		if dispatchErr != nil {
+			if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, monitorDesktopTaskSpecIDs(phase2TaskSpecs)); fallbackErr != nil {
+				return nil, fallbackErr
+			}
+			phase2DispatchReady = false
+			phase2TaskSpecs = nil
+			if s.logger != nil {
+				s.logger.Warn("phase2 monitor desktop dispatch failed, falling back to legacy execution",
+					zap.Error(dispatchErr),
+					zap.Int("phase2_task_count", phase2TaskCount),
+				)
+			}
+		} else {
+			if len(fallbackLegacyTaskIDs) > 0 {
+				if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, fallbackLegacyTaskIDs); fallbackErr != nil {
+					return nil, fallbackErr
+				}
+			}
+			phase2PublishedTasks = publishedTasks
+		}
 	}

-	return &MonitoringCollectNowResponse{
-		CollectionMode:       quota.CollectionMode,
-		RefreshedTaskCount:   refreshedCount,
-		CreatedTaskCount:     createdCount,
-		LeasedTaskCount:      leasedCount,
-		CompletedTaskCount:   completedCount,
-		HasEffectiveSnapshot: hasEffectiveSnapshot,
-		Message:              message,
-	}, nil
+	phase2InterruptTargets := make([]monitorDesktopInterruptTarget, 0)
+	phase2DeferredTasks := make([]*repository.DesktopTask, 0)
+	if options.Preempt && phase2DispatchReady {
+		excludedMonitorTaskIDs := make([]int64, 0, len(phase2TaskSpecs))
+		for _, spec := range phase2TaskSpecs {
+			excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID)
+		}
+		phase2DeferredTasks, err = s.deferQueuedPhase2MonitorTasks(ctx, *clientID, excludedMonitorTaskIDs)
+		if err != nil {
+			return nil, err
+		}
+		if len(phase2DeferredTasks) > 0 {
+			abortedQueuedCount += int64(len(phase2DeferredTasks))
+			if _, updateErr := s.monitoringPool.Exec(ctx, `
+				UPDATE monitoring_collect_requests
+				SET aborted_queued_count = aborted_queued_count + $2,
+				    updated_at = NOW()
+				WHERE request_id = $1
+			`, requestID, len(phase2DeferredTasks)); updateErr != nil {
+				return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 deferred queue count")
+			}
+		}
+		phase2InterruptTargets, err = s.requestPhase2MonitorInterrupts(ctx, *clientID, excludedMonitorTaskIDs, interruptGeneration)
+		if err != nil {
+			return nil, err
+		}
+		if len(phase2InterruptTargets) > 0 {
+			if _, updateErr := s.monitoringPool.Exec(ctx, `
+				UPDATE monitoring_collect_requests
+				SET interrupt_requested_count = interrupt_requested_count + $2,
+				    updated_at = NOW()
+				WHERE request_id = $1
+			`, requestID, len(phase2InterruptTargets)); updateErr != nil {
+				return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 interrupt request count")
+			}
+		}
+	}
+
+	for _, task := range phase2PublishedTasks {
+		if publishErr := publishMonitorDesktopTaskAvailable(
+			context.Background(),
+			task.TargetClientID.String(),
+			task,
+			func(publishCtx context.Context, event DesktopTaskEvent) error {
+				return publishDesktopTaskEvent(publishCtx, s.rabbitMQ, event)
+			},
+			func(publishCtx context.Context, event stream.DesktopDispatchEvent) error {
+				return publishDesktopDispatchEvent(publishCtx, s.rabbitMQ, s.logger, event)
+			},
+		); publishErr != nil && s.logger != nil {
+			s.logger.Warn("phase2 monitor desktop task available publish failed")
+		}
+	}
+	for _, task := range phase2DeferredTasks {
+		if publishErr := publishMonitorDesktopTaskAvailable(
+			context.Background(),
+			task.TargetClientID.String(),
+			task,
+			func(publishCtx context.Context, event DesktopTaskEvent) error {
+				return publishDesktopTaskEvent(publishCtx, s.rabbitMQ, event)
+			},
+			func(publishCtx context.Context, event stream.DesktopDispatchEvent) error {
+				return publishDesktopDispatchEvent(publishCtx, s.rabbitMQ, s.logger, event)
+			},
+		); publishErr != nil && s.logger != nil {
+			s.logger.Warn("phase2 deferred monitor desktop task publish failed",
+				zap.String("task_id", task.DesktopID.String()),
+				zap.Error(publishErr),
+			)
+		}
+	}
+	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,
+		CreatedTaskCount:        createdCount,
+		LeasedTaskCount:         leasedCount,
+		CompletedTaskCount:      completedCount,
+		HasEffectiveSnapshot:    hasEffectiveSnapshot,
+		RequestID:               requestID.String(),
+		TargetClientID:          clientID.String(),
+		AffectedTaskCount:       affectedTaskCount,
+		PromotedTaskCount:       affectedTaskCount,
+		AbortedQueuedCount:      abortedQueuedCount,
+		InterruptRequestedCount: totalInterruptRequestedCount,
+		InterruptGeneration:     interruptGeneration,
+		TTLExpiresAt:            ttlExpiresAt.Format(time.RFC3339),
+		Message:                 message,
+	}
+	return s.finalizeCollectNowResponse(ctx, requestID, quota.CollectionMode, options.WaitForFirstDispatch, responseData), nil
 }

-func (s *MonitoringService) resolveCollectNowClientID(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) {
+func (s *MonitoringService) resolveCollectNowClientID(
+	ctx context.Context,
+	tenantID, workspaceID, userID int64,
+	preferredClientID *uuid.UUID,
+) (*uuid.UUID, error) {
+	if preferredClientID != nil {
+		if err := s.ensureClientBelongsToUser(ctx, tenantID, workspaceID, userID, *preferredClientID); err != nil {
+			return nil, err
+		}
+		return preferredClientID, nil
+	}
 	return s.findLatestOnlineClientForUser(ctx, tenantID, workspaceID, userID)
 }

@@ -623,6 +913,97 @@ func (s *MonitoringService) hasOnlineClientForUser(ctx context.Context, tenantID
 	return clientID != nil, nil
 }

+func (s *MonitoringService) ensureClientBelongsToUser(
+	ctx context.Context,
+	tenantID, workspaceID, userID int64,
+	clientID uuid.UUID,
+) error {
+	var exists bool
+	if err := s.businessPool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1
+			FROM desktop_clients
+			WHERE id = $1
+			  AND tenant_id = $2
+			  AND workspace_id = $3
+			  AND user_id = $4
+			  AND revoked_at IS NULL
+			  AND last_seen_at IS NOT NULL
+			  AND last_seen_at >= NOW() - $5::interval
+		)
+	`, clientID, tenantID, workspaceID, userID, formatPgInterval(desktopClientPresenceTTL)).Scan(&exists); err != nil {
+		return response.ErrInternal(50041, "query_failed", "failed to validate target monitoring desktop client")
+	}
+	if !exists {
+		return response.ErrConflict(40941, "desktop_client_offline", "current user's monitoring desktop client is offline")
+	}
+	return nil
+}
+
+func (s *MonitoringService) nextCollectNowInterruptGeneration(
+	ctx context.Context,
+	tx pgx.Tx,
+	clientID uuid.UUID,
+	executionOwner string,
+) (int, error) {
+	var next int
+	if strings.TrimSpace(executionOwner) == "desktop_tasks" && s.businessPool != nil {
+		if err := s.businessPool.QueryRow(ctx, `
+			SELECT COALESCE(MAX(interrupt_generation), 0) + 1
+			FROM desktop_tasks
+			WHERE kind = 'monitor'
+			  AND target_client_id = $1
+		`, clientID).Scan(&next); err != nil {
+			return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation")
+		}
+	} else {
+		if err := tx.QueryRow(ctx, `
+			SELECT COALESCE(MAX(interrupt_generation), 0) + 1
+			FROM monitoring_collect_tasks
+			WHERE target_client_id = $1
+		`, clientID).Scan(&next); err != nil {
+			return 0, response.ErrInternal(50041, "query_failed", "failed to allocate collect-now interrupt generation")
+		}
+	}
+	if next <= 0 {
+		next = 1
+	}
+	return next, nil
+}
+
+func monitoringPlatformIDs(items []monitoringPlatformMetadata) []string {
+	if len(items) == 0 {
+		return nil
+	}
+	result := make([]string, 0, len(items))
+	for _, item := range items {
+		if strings.TrimSpace(item.ID) == "" {
+			continue
+		}
+		result = append(result, item.ID)
+	}
+	return result
+}
+
+func hashMonitoringCollectNowScope(
+	userID int64,
+	brandID int64,
+	keywordID *int64,
+	questionIDs []int64,
+	platformIDs []string,
+) string {
+	payload := map[string]any{
+		"user_id":      userID,
+		"brand_id":     brandID,
+		"keyword_id":   keywordID,
+		"question_ids": questionIDs,
+		"platform_ids": platformIDs,
+	}
+	body, _ := json.Marshal(payload)
+	sum := md5.Sum(body)
+	return hex.EncodeToString(sum[:])
+}
+
 func (s *MonitoringService) findLatestOnlineClientForUser(ctx context.Context, tenantID, workspaceID, userID int64) (*uuid.UUID, error) {
 	var clientID uuid.UUID
 	err := s.businessPool.QueryRow(ctx, `
@@ -836,12 +1217,19 @@ func (s *MonitoringService) ensureCollectNowTasks(
 	questions []monitoringConfiguredQuestion,
 	platforms []monitoringPlatformMetadata,
 	businessDate time.Time,
-) (int64, int64, error) {
+	targetClientID uuid.UUID,
+	interruptGeneration int,
+	executionOwner string,
+) (int64, int64, int64, int64, []int64, error) {
 	const runMode = "plugin_standard"
+	if strings.TrimSpace(executionOwner) == "" {
+		executionOwner = "legacy"
+	}

 	dateText := businessDate.Format("2006-01-02")
 	var refreshedCount int64
 	var createdCount int64
+	interruptTaskIDs := make([]int64, 0)

 	for _, question := range questions {
 		for _, platform := range platforms {
@@ -851,6 +1239,18 @@ func (s *MonitoringService) ensureCollectNowTasks(
 				    status = 'pending',
 				    planned_at = NOW(),
 				    trigger_source = 'manual',
+				    dispatch_priority = 5000,
+				    dispatch_lane = 'high',
+				    target_client_id = $9,
+				    dispatch_after = NOW(),
+				    interrupt_generation = $10,
+				    superseded_by_request_id = NULL,
+				    last_dispatched_at = NULL,
+				    execution_owner = $11,
+				    ingest_shard_key = COALESCE(
+				        NULLIF(ingest_shard_key, ''),
+				        tenant_id::text || ':' || brand_id::text || ':' || business_date::text
+				    ),
 				    lease_token_hash = NULL,
 				    leased_to_executor = NULL,
 				    leased_at = NULL,
@@ -869,34 +1269,139 @@ func (s *MonitoringService) ensureCollectNowTasks(
 				  AND run_mode = $6
 				  AND business_date = $8::date
 				  AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped')
-			`, tenantID, brandID, question.ID, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText)
+			`, tenantID, brandID, question.ID, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, interruptGeneration, executionOwner)
 			if err != nil {
-				return 0, 0, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
+				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
 			}
 			refreshedCount += tag.RowsAffected()

+			rows, err := tx.Query(ctx, `
+				UPDATE monitoring_collect_tasks
+				SET question_hash = $7,
+				    trigger_source = 'manual',
+				    dispatch_priority = 5000,
+				    dispatch_lane = 'high',
+				    target_client_id = $9,
+				    dispatch_after = NOW(),
+				    interrupt_generation = $10,
+				    superseded_by_request_id = NULL,
+				    last_dispatched_at = NULL,
+				    execution_owner = COALESCE(execution_owner, 'legacy'),
+				    ingest_shard_key = COALESCE(
+				        NULLIF(ingest_shard_key, ''),
+				        tenant_id::text || ':' || brand_id::text || ':' || business_date::text
+				    ),
+				    updated_at = NOW()
+				WHERE tenant_id = $1
+				  AND brand_id = $2
+				  AND question_id = $3
+				  AND ai_platform_id = $4
+				  AND collector_type = $5
+				  AND run_mode = $6
+				  AND business_date = $8::date
+				  AND status = 'leased'
+				  AND COALESCE(execution_owner, 'legacy') = 'legacy'
+				RETURNING id
+			`, tenantID, brandID, question.ID, platform.ID, monitoringCollectorType, runMode, question.QuestionHash, dateText, targetClientID, interruptGeneration)
+			if err != nil {
+				return 0, 0, 0, 0, nil, 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 (
 					tenant_id, brand_id, question_id, question_hash, ai_platform_id,
-					collector_type, trigger_source, run_mode, business_date, planned_at, status
+					collector_type, trigger_source, run_mode, business_date, planned_at, status,
+					dispatch_priority, dispatch_lane, target_client_id, dispatch_after,
+					interrupt_generation, ingest_shard_key, execution_owner
 				)
 				VALUES (
 					$1, $2, $3, $4, $5,
-					$6, 'manual', $7, $8::date, NOW(), 'pending'
+					$6, 'manual', $7, $8::date, NOW(), 'pending',
+					5000, 'high', $9, NOW(),
+					$10, $11, $12
 				)
 				ON CONFLICT (
 					tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
 				)
 				DO NOTHING
-			`, tenantID, brandID, question.ID, question.QuestionHash, platform.ID, monitoringCollectorType, runMode, dateText)
+			`, tenantID, brandID, question.ID, question.QuestionHash, platform.ID, monitoringCollectorType, runMode, dateText, targetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner)
 			if err != nil {
-				return 0, 0, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks")
+				return 0, 0, 0, 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks")
 			}
 			createdCount += tag.RowsAffected()
 		}
 	}

-	return refreshedCount, createdCount, nil
+	var leasedCount int64
+	var completedCount int64
+	if err := tx.QueryRow(ctx, `
+		SELECT
+			COUNT(*) FILTER (WHERE status = 'leased') AS leased_count,
+			COUNT(*) FILTER (WHERE status = 'completed') AS completed_count
+		FROM monitoring_collect_tasks
+		WHERE tenant_id = $1
+		  AND brand_id = $2
+		  AND collector_type = $3
+		  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 refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, nil
+}
+
+func (s *MonitoringService) deferQueuedNormalTasks(
+	ctx context.Context,
+	tx pgx.Tx,
+	tenantID int64,
+	businessDate time.Time,
+	brandID int64,
+	questionIDs []int64,
+	platformIDs []string,
+	targetClientID uuid.UUID,
+	requestID uuid.UUID,
+	ttlExpiresAt time.Time,
+) (int64, error) {
+	if tx == nil || len(questionIDs) == 0 || len(platformIDs) == 0 {
+		return 0, nil
+	}
+
+	tag, err := tx.Exec(ctx, `
+		UPDATE monitoring_collect_tasks
+		SET dispatch_after = GREATEST(COALESCE(dispatch_after, NOW()), $6),
+		    superseded_by_request_id = $5,
+		    updated_at = NOW()
+		WHERE tenant_id = $1
+		  AND collector_type = $2
+		  AND business_date = $3::date
+		  AND COALESCE(execution_owner, 'legacy') = 'legacy'
+		  AND status = 'pending'
+		  AND target_client_id = $4
+		  AND dispatch_lane IN ('normal', 'normal_boosted', 'retry')
+		  AND NOT (
+			brand_id = $7
+			AND question_id = ANY($8::bigint[])
+			AND ai_platform_id = ANY($9::text[])
+		  )
+	`, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), targetClientID, requestID, ttlExpiresAt, brandID, questionIDs, platformIDs)
+	if err != nil {
+		return 0, response.ErrInternal(50041, "update_failed", "failed to defer queued normal monitoring tasks")
+	}
+	return tag.RowsAffected(), nil
 }

 func (s *MonitoringService) loadQuota(ctx context.Context, tenantID int64) (monitoringQuotaConfig, error) {
@@ -2633,6 +3138,40 @@ func resolveMonitoringPlatforms(enabled []string) []monitoringPlatformMetadata {
 	return platforms
 }

+func intersectMonitoringPlatforms(enabled []monitoringPlatformMetadata, requested []string) ([]monitoringPlatformMetadata, error) {
+	if len(requested) == 0 {
+		return enabled, nil
+	}
+
+	enabledIndex := make(map[string]monitoringPlatformMetadata, len(enabled))
+	for _, platform := range enabled {
+		enabledIndex[platform.ID] = platform
+	}
+
+	result := make([]monitoringPlatformMetadata, 0, len(requested))
+	seen := make(map[string]struct{}, len(requested))
+	for _, raw := range requested {
+		id := strings.TrimSpace(raw)
+		if id == "" {
+			continue
+		}
+		if _, duplicate := seen[id]; duplicate {
+			continue
+		}
+		seen[id] = struct{}{}
+		platform, ok := enabledIndex[id]
+		if !ok {
+			return nil, response.ErrBadRequest(40031, "platform_not_enabled", fmt.Sprintf("platform %s is not enabled for this tenant", id))
+		}
+		result = append(result, platform)
+	}
+
+	if len(result) == 0 {
+		return nil, response.ErrBadRequest(40031, "invalid_platform_ids", "platform_ids contains no usable platform")
+	}
+	return result, nil
+}
+
 func filterMonitoringPlatforms(platforms []monitoringPlatformMetadata, aiPlatformID *string) []monitoringPlatformMetadata {
 	platformID := normalizedOptionalString(aiPlatformID)
 	if platformID == "" {