enable desktop monitoring rollout

deploy/config.yaml

@@ -95,7 +95,7 @@ monitoring_workers:
   projection_rebuild_concurrency: 2
 
 monitoring_dispatch:
-  desktop_tasks_rollout_percentage: 5
+  desktop_tasks_rollout_percentage: 100
 
 membership:
   default_plan_code: free

deploy/k3s/config/config.yaml

@@ -100,7 +100,7 @@ monitoring_workers:
   projection_rebuild_concurrency: 2
 
 monitoring_dispatch:
-  desktop_tasks_rollout_percentage: 5
+  desktop_tasks_rollout_percentage: 100
 
 membership:
   default_plan_code: free

server/cmd/scheduler/main.go

@@ -344,20 +344,9 @@ func shutdownSchedulerMetricsServer(server *http.Server, logger interface {
 }
 
 func monitoringDailyRunOptions(jobCtx internalscheduler.JobRunContext) tenantapp.MonitoringDailyTaskRunOptions {
-	batchSize := 0
-	if jobCtx.Job != nil && jobCtx.Job.BatchSize != nil {
-		batchSize = *jobCtx.Job.BatchSize
-	}
-	if batchSize <= 0 {
-		batchSize = internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_run", 0)
-	}
-	if batchSize <= 0 {
-		batchSize = internalscheduler.AsInt(jobCtx.Config, "batch_size", 0)
-	}
 	return tenantapp.MonitoringDailyTaskRunOptions{
-		MaxMaterializePerRun:   batchSize,
+		MaxDesktopDispatchPerRun: internalscheduler.AsInt(jobCtx.Config, "max_desktop_dispatch_per_run", 0),
-		MaxMaterializePerPlan:  internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_plan", 0),
+		MaxDesktopClientBacklog:  internalscheduler.AsInt(jobCtx.Config, "max_desktop_client_backlog", 0),
-		MaxMaterializePerBrand: internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_brand", 0),
 	}
 }
 

server/cmd/scheduler/main_test.go

@@ -59,22 +59,30 @@ func TestSchedulerMetricsAuthMiddleware(t *testing.T) {
 	}
 }
 
-func TestMonitoringDailyRunOptionsUsesJobBatchSize(t *testing.T) {
+func TestMonitoringDailyRunOptionsIgnoresJobBatchSizeForDailyCollection(t *testing.T) {
 	batchSize := 88
 	got := monitoringDailyRunOptions(internalscheduler.JobRunContext{
 		Job: &opsdomain.SchedulerJob{BatchSize: &batchSize},
 		Config: map[string]any{
-			"max_materialize_per_run":   999,
+			"max_materialize_per_run":    999,
-			"max_materialize_per_plan":  12,
+			"max_materialize_per_plan":   12,
-			"max_materialize_per_brand": 4,
+			"max_materialize_per_brand":  4,
+			"max_desktop_client_backlog": 6,
+			"batch_size":                 77,
 		},
 	})
 
-	if got.MaxMaterializePerRun != 88 {
+	if got.MaxDesktopDispatchPerRun != 0 {
-		t.Fatalf("MaxMaterializePerRun = %d, want 88", got.MaxMaterializePerRun)
+		t.Fatalf("MaxDesktopDispatchPerRun = %d, want unlimited", got.MaxDesktopDispatchPerRun)
 	}
-	if got.MaxMaterializePerPlan != 12 || got.MaxMaterializePerBrand != 4 {
+	if got.MaxMaterializePerPlan != 0 || got.MaxMaterializePerBrand != 0 {
-		t.Fatalf("unexpected plan/brand options: %#v", got)
+		t.Fatalf("legacy plan/brand throttles should be ignored: %#v", got)
+	}
+	if got.MaxMaterializePerRun != 0 {
+		t.Fatalf("batch_size must not throttle daily materialization: %#v", got)
+	}
+	if got.MaxDesktopClientBacklog != 6 {
+		t.Fatalf("MaxDesktopClientBacklog = %d, want 6", got.MaxDesktopClientBacklog)
 	}
 }
 

server/configs/config.yaml

@@ -100,7 +100,7 @@ monitoring_workers:
   projection_rebuild_concurrency: 2
 
 monitoring_dispatch:
-  desktop_tasks_rollout_percentage: 5
+  desktop_tasks_rollout_percentage: 100
 
 membership:
   default_plan_code: free

server/internal/tenant/app/monitoring_callback_service.go

@@ -40,7 +40,9 @@ const (
 	// reports skip writes and only refresh the row at this interval so dashboards
 	// do not depend on per-heartbeat database churn.
 	monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
-	monitoringRuntimeUnknownMaxDispatchAttempts   = 3
+	monitoringRuntimeUnknownMaxDispatchAttempts   = 2
+	monitoringAutomaticFailureMaxAttempts         = 2
+	monitoringAutomaticFailureRetryAfter          = 2 * time.Minute
 )
 
 var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`)
@@ -240,27 +242,28 @@ type monitoringHeartbeatSnapshotDecision struct {
 }
 
 type monitoringCollectTask struct {
-	ID                 int64
+	ID                    int64
-	TenantID           int64
+	TenantID              int64
-	BrandID            int64
+	BrandID               int64
-	QuestionID         int64
+	QuestionID            int64
-	QuestionHash       []byte
+	QuestionHash          []byte
-	AIPlatformID       string
+	AIPlatformID          string
-	CollectorType      string
+	CollectorType         string
-	RunMode            string
+	RunMode               string
-	BusinessDate       time.Time
+	BusinessDate          time.Time
-	Status             string
+	Status                string
-	ExecutionOwner     string
+	ExecutionOwner        string
-	LeaseTokenHash     sql.NullString
+	LeaseTokenHash        sql.NullString
-	LeasedToExecutor   sql.NullString
+	LeasedToExecutor      sql.NullString
-	LeaseExpiresAt     sql.NullTime
+	LeaseExpiresAt        sql.NullTime
-	CallbackReceivedAt sql.NullTime
+	CallbackReceivedAt    sql.NullTime
-	CompletedAt        sql.NullTime
+	CompletedAt           sql.NullTime
-	SkipReason         sql.NullString
+	SkipReason            sql.NullString
-	TriggerSource      string
+	TriggerSource         string
-	QuestionText       string
+	QuestionText          string
-	TargetClientID     sql.NullString
+	TargetClientID        sql.NullString
-	DispatchAttempts   int
+	DispatchAttempts      int
+	AutomaticFailureCount int
 }
 
 type monitoringDesktopExecutionLease struct {
@@ -1062,6 +1065,32 @@ type monitoringSkipOutcome struct {
 	RetryAfter       time.Duration
 }
 
+type monitoringFailedResultOutcome struct {
+	TaskStatus string
+	Requeue    bool
+	RetryAfter time.Duration
+}
+
+func monitoringFailedResultOutcomeForTask(task *monitoringCollectTask, req MonitoringTaskResultRequest) monitoringFailedResultOutcome {
+	if task == nil {
+		return monitoringFailedResultOutcome{TaskStatus: "failed"}
+	}
+	if !strings.EqualFold(strings.TrimSpace(task.TriggerSource), "automatic") {
+		return monitoringFailedResultOutcome{TaskStatus: "failed"}
+	}
+	if monitoringTaskFailureDispositionFromRequest(req).NonRetryable {
+		return monitoringFailedResultOutcome{TaskStatus: "failed"}
+	}
+	if task.AutomaticFailureCount < monitoringAutomaticFailureMaxAttempts-1 {
+		return monitoringFailedResultOutcome{
+			TaskStatus: "pending",
+			Requeue:    true,
+			RetryAfter: monitoringAutomaticFailureRetryAfter,
+		}
+	}
+	return monitoringFailedResultOutcome{TaskStatus: "failed"}
+}
+
 func monitoringSkipOutcomeForReason(skipReason string, dispatchAttempts int, triggerSource string) monitoringSkipOutcome {
 	switch strings.TrimSpace(skipReason) {
 	case "runtime_unknown":
@@ -1562,6 +1591,11 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 			trigger_source,
 			COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
 			COALESCE(target_client_id::text, '') AS target_client_id,
+			CASE
+				WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
+					THEN (t.request_payload_json ->> 'automatic_failure_count')::int
+				ELSE 0
+			END AS automatic_failure_count,
 			COALESCE(NULLIF(t.question_text_snapshot, ''), (
 				SELECT question_text_snapshot
 				FROM monitoring_question_config_snapshots
@@ -1600,6 +1634,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 		&item.TriggerSource,
 		&item.DispatchAttempts,
 		&item.TargetClientID,
+		&item.AutomaticFailureCount,
 		&item.QuestionText,
 	); err != nil {
 		if err == pgx.ErrNoRows {
@@ -1635,6 +1670,11 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 			trigger_source,
 			COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
 			COALESCE(target_client_id::text, '') AS target_client_id,
+			CASE
+				WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
+					THEN (t.request_payload_json ->> 'automatic_failure_count')::int
+				ELSE 0
+			END AS automatic_failure_count,
 			COALESCE(NULLIF(t.question_text_snapshot, ''), (
 				SELECT question_text_snapshot
 				FROM monitoring_question_config_snapshots
@@ -1672,6 +1712,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 		&item.TriggerSource,
 		&item.DispatchAttempts,
 		&item.TargetClientID,
+		&item.AutomaticFailureCount,
 		&item.QuestionText,
 	); err != nil {
 		if err == pgx.ErrNoRows {
@@ -2312,12 +2353,20 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
 	}
 	defer tx.Rollback(ctx)
 
-	runID, err := s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
+	var failedOutcome monitoringFailedResultOutcome
-	if err != nil {
+	if runStatus == "failed" {
-		return nil, err
+		failedOutcome = monitoringFailedResultOutcomeForTask(task, req)
 	}
 
-	if runStatus == "succeeded" {
+	var runID int64
+	if !failedOutcome.Requeue {
+		runID, err = s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if runStatus == "succeeded" && !failedOutcome.Requeue {
 		if err := s.upsertParseResult(ctx, tx, task, runID, parseReq, parseVersion); err != nil {
 			return nil, err
 		}
@@ -2325,10 +2374,12 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
 
 	citationSourceCount := 0
 	contentCitationCount := 0
-	if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
+	if !failedOutcome.Requeue {
-		return nil, err
+		if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
+			return nil, err
+		}
 	}
-	if runStatus == "succeeded" {
+	if runStatus == "succeeded" && !failedOutcome.Requeue {
 		facts, buildErr := s.buildCitationFacts(ctx, tx, task.TenantID, task.AIPlatformID, sourceInputs)
 		if buildErr != nil {
 			return nil, buildErr
@@ -2346,12 +2397,16 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
 
 	taskStatus := "completed"
 	if runStatus == "failed" {
-		taskStatus = "failed"
+		taskStatus = failedOutcome.TaskStatus
 	}
-	if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
+	if failedOutcome.Requeue {
+		if err := s.requeueCollectTaskAfterFailedResult(ctx, tx, task.ID, failedOutcome.RetryAfter, req.ErrorMessage); err != nil {
+			return nil, err
+		}
+	} else if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
 		return nil, err
 	}
-	if runStatus == "failed" {
+	if runStatus == "failed" && !failedOutcome.Requeue {
 		if _, err := s.failRemainingMonitoringTasksForAuthorizationFailure(ctx, tx, task, req, completedAt); err != nil {
 			return nil, err
 		}
@@ -2365,9 +2420,13 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
 		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest")
 	}
 
+	var runIDPtr *int64
+	if !failedOutcome.Requeue {
+		runIDPtr = &runID
+	}
 	return &MonitoringTaskResultResponse{
 		TaskID:               task.ID,
-		RunID:                &runID,
+		RunID:                runIDPtr,
 		TaskStatus:           taskStatus,
 		CitationSourceCount:  citationSourceCount,
 		ContentCitationCount: contentCitationCount,
@@ -3113,6 +3172,51 @@ func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context,
 	return nil
 }
 
+func (s *MonitoringCallbackService) requeueCollectTaskAfterFailedResult(ctx context.Context, tx pgx.Tx, taskID int64, retryAfter time.Duration, errorMessage *string) error {
+	if retryAfter <= 0 {
+		retryAfter = monitoringAutomaticFailureRetryAfter
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE monitoring_collect_tasks
+		SET status = 'pending',
+		    lease_token_hash = NULL,
+		    leased_to_executor = NULL,
+		    leased_at = NULL,
+		    lease_expires_at = NULL,
+		    callback_received_at = NULL,
+		    completed_at = NULL,
+		    skip_reason = NULL,
+		    last_dispatched_at = NULL,
+		    dispatch_after = NOW() + $2::interval,
+		    error_message = $3,
+		    request_payload_json = (
+				CASE
+					WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
+						THEN '{}'::jsonb
+					ELSE request_payload_json
+				END
+			) || jsonb_build_object(
+				'automatic_failure_count',
+				(
+					CASE
+						WHEN COALESCE(request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
+							THEN (request_payload_json ->> 'automatic_failure_count')::int
+						ELSE 0
+					END
+				) + 1,
+				'automatic_failure_retry_after_seconds',
+				$4::int,
+				'automatic_failure_last_requeued_at',
+				to_jsonb(NOW())
+			),
+		    updated_at = NOW()
+		WHERE id = $1
+	`, taskID, formatPgInterval(retryAfter), nullableString(errorMessage), int(retryAfter.Seconds())); err != nil {
+		return response.ErrInternal(50041, "task_update_failed", "failed to requeue monitoring task after failed result")
+	}
+	return nil
+}
+
 func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorizationFailure(
 	ctx context.Context,
 	tx pgx.Tx,

server/internal/tenant/app/monitoring_callback_service_test.go

@@ -499,6 +499,73 @@ func TestMonitoringTaskFailureDispositionDetectsAuthorizationFailureFromMessage(
 	}
 }
 
+func TestMonitoringFailedResultOutcomeRequeuesAutomaticFirstFailure(t *testing.T) {
+	message := "模型调用超时"
+	got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{
+		TriggerSource:         "automatic",
+		AutomaticFailureCount: 0,
+	}, MonitoringTaskResultRequest{
+		Status:       "failed",
+		ErrorMessage: &message,
+	})
+
+	if got.TaskStatus != "pending" {
+		t.Fatalf("TaskStatus = %q, want pending", got.TaskStatus)
+	}
+	if !got.Requeue {
+		t.Fatal("Requeue = false, want true")
+	}
+	if got.RetryAfter != monitoringAutomaticFailureRetryAfter {
+		t.Fatalf("RetryAfter = %v, want %v", got.RetryAfter, monitoringAutomaticFailureRetryAfter)
+	}
+}
+
+func TestMonitoringFailedResultOutcomeFailsAutomaticSecondFailure(t *testing.T) {
+	got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{
+		TriggerSource:         "automatic",
+		AutomaticFailureCount: monitoringAutomaticFailureMaxAttempts - 1,
+	}, MonitoringTaskResultRequest{Status: "failed"})
+
+	if got.TaskStatus != "failed" {
+		t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus)
+	}
+	if got.Requeue {
+		t.Fatal("Requeue = true, want false")
+	}
+}
+
+func TestMonitoringFailedResultOutcomeDoesNotRetryAuthorizationFailure(t *testing.T) {
+	message := "登录态已失效，请重新授权后再试。"
+	got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{
+		TriggerSource:         "automatic",
+		AutomaticFailureCount: 0,
+	}, MonitoringTaskResultRequest{
+		Status:       "failed",
+		ErrorMessage: &message,
+	})
+
+	if got.TaskStatus != "failed" {
+		t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus)
+	}
+	if got.Requeue {
+		t.Fatal("Requeue = true, want false")
+	}
+}
+
+func TestMonitoringFailedResultOutcomeDoesNotRetryManualTask(t *testing.T) {
+	got := monitoringFailedResultOutcomeForTask(&monitoringCollectTask{
+		TriggerSource:         "manual",
+		AutomaticFailureCount: 0,
+	}, MonitoringTaskResultRequest{Status: "failed"})
+
+	if got.TaskStatus != "failed" {
+		t.Fatalf("TaskStatus = %q, want failed", got.TaskStatus)
+	}
+	if got.Requeue {
+		t.Fatal("Requeue = true, want false")
+	}
+}
+
 func TestMonitoringAuthorizationFailureTargetClientPrefersExplicitTargetClient(t *testing.T) {
 	targetClientID := uuid.New().String()
 	task := &monitoringCollectTask{

server/internal/tenant/app/monitoring_daily_task_metrics.go

@@ -20,6 +20,7 @@ type MonitoringDailyTaskMetricsSnapshot struct {
 	BrandFailureTotal      int64      `json:"brand_failure_total"`
 	SkippedPlanTotal       int64      `json:"skipped_plan_total"`
 	PlannedTaskTotal       int64      `json:"planned_task_total"`
+	EligibleTaskTotal      int64      `json:"eligible_task_total"`
 	DueTaskTotal           int64      `json:"due_task_total"`
 	CreatedTaskTotal       int64      `json:"created_task_total"`
 	DesktopTaskTotal       int64      `json:"desktop_task_total"`
@@ -44,6 +45,7 @@ type monitoringDailyTaskMetricsRecorder struct {
 	brandFailureTotal      atomic.Int64
 	skippedPlanTotal       atomic.Int64
 	plannedTaskTotal       atomic.Int64
+	eligibleTaskTotal      atomic.Int64
 	dueTaskTotal           atomic.Int64
 	createdTaskTotal       atomic.Int64
 	desktopTaskTotal       atomic.Int64
@@ -98,6 +100,7 @@ func (r *monitoringDailyTaskMetricsRecorder) record(summary *MonitoringDailyTask
 		r.brandFailureTotal.Add(int64(summary.FailedBrandCount))
 		r.skippedPlanTotal.Add(int64(summary.SkippedPlanCount))
 		r.plannedTaskTotal.Add(summary.PlannedTaskCount)
+		r.eligibleTaskTotal.Add(summary.EligibleTaskCount)
 		r.dueTaskTotal.Add(summary.DueTaskCount)
 		r.createdTaskTotal.Add(summary.CreatedTaskCount)
 		r.desktopTaskTotal.Add(summary.DesktopTaskCount)
@@ -133,6 +136,7 @@ func (r *monitoringDailyTaskMetricsRecorder) snapshot() MonitoringDailyTaskMetri
 		BrandFailureTotal:      r.brandFailureTotal.Load(),
 		SkippedPlanTotal:       r.skippedPlanTotal.Load(),
 		PlannedTaskTotal:       r.plannedTaskTotal.Load(),
+		EligibleTaskTotal:      r.eligibleTaskTotal.Load(),
 		DueTaskTotal:           r.dueTaskTotal.Load(),
 		CreatedTaskTotal:       r.createdTaskTotal.Load(),
 		DesktopTaskTotal:       r.desktopTaskTotal.Load(),

server/internal/tenant/app/monitoring_daily_task_worker.go

@@ -19,19 +19,20 @@ import (
 )
 
 const (
-	defaultMonitoringDailyTaskInterval           = 5 * time.Minute
+	defaultMonitoringDailyTaskInterval            = 5 * time.Minute
-	defaultMonitoringDailyTaskTimeout            = 45 * time.Second
+	defaultMonitoringDailyTaskTimeout             = 45 * time.Second
-	defaultMonitoringDailyTaskHardCap            = 36
+	defaultMonitoringDailyTaskHardCap             = 36
-	defaultMonitoringDailyBrandLimit             = 1
+	defaultMonitoringDailyBrandLimit              = 1
-	defaultMonitoringDailyLookahead              = 10 * time.Minute
+	defaultMonitoringDailyLookahead               = 10 * time.Minute
-	defaultMonitoringDailyMaterializeWindow      = 2 * time.Hour
+	defaultMonitoringDailyDesktopRetryCooldown    = 15 * time.Minute
-	defaultMonitoringDailyDesktopRetryCooldown   = 15 * time.Minute
+	defaultMonitoringDailyPublishTimeout          = 5 * time.Second
-	defaultMonitoringDailyPublishTimeout         = 5 * time.Second
+	defaultMonitoringDailyMaxMaterializePerRun    = 0
-	defaultMonitoringDailyMaxMaterializePerRun   = 64
+	defaultMonitoringDailyMaxMaterializePerPlan   = 0
-	defaultMonitoringDailyMaxMaterializePerPlan  = 12
+	defaultMonitoringDailyMaxMaterializePerBrand  = 0
-	defaultMonitoringDailyMaxMaterializePerBrand = 4
+	defaultMonitoringDailyMaxDesktopDispatch      = 0
-	monitoringDailyPlatformActiveWindow          = 24 * time.Hour
+	defaultMonitoringDailyMaxDesktopClientBacklog = 12
-	monitoringDailyMaterializeStartOffset        = 5 * time.Minute
+	monitoringDailyPlatformActiveWindow           = 24 * time.Hour
+	monitoringDailyMaterializeStartOffset         = 5 * time.Minute
 )
 
 type MonitoringDailyTaskWorker struct {
@@ -49,6 +50,7 @@ type MonitoringDailyTaskGenerationSummary struct {
 	FailedPlanCount   int
 	FailedBrandCount  int
 	PlannedTaskCount  int64
+	EligibleTaskCount int64
 	DueTaskCount      int64
 	CreatedTaskCount  int64
 	DesktopTaskCount  int64
@@ -57,16 +59,20 @@ type MonitoringDailyTaskGenerationSummary struct {
 }
 
 type MonitoringDailyTaskRunOptions struct {
-	MaxMaterializePerRun   int
+	MaxMaterializePerRun     int
-	MaxMaterializePerPlan  int
+	MaxMaterializePerPlan    int
-	MaxMaterializePerBrand int
+	MaxMaterializePerBrand   int
+	MaxDesktopDispatchPerRun int
+	MaxDesktopClientBacklog  int
 }
 
 func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions) MonitoringDailyTaskRunOptions {
 	out := MonitoringDailyTaskRunOptions{
-		MaxMaterializePerRun:   defaultMonitoringDailyMaxMaterializePerRun,
+		MaxMaterializePerRun:     defaultMonitoringDailyMaxMaterializePerRun,
-		MaxMaterializePerPlan:  defaultMonitoringDailyMaxMaterializePerPlan,
+		MaxMaterializePerPlan:    defaultMonitoringDailyMaxMaterializePerPlan,
-		MaxMaterializePerBrand: defaultMonitoringDailyMaxMaterializePerBrand,
+		MaxMaterializePerBrand:   defaultMonitoringDailyMaxMaterializePerBrand,
+		MaxDesktopDispatchPerRun: defaultMonitoringDailyMaxDesktopDispatch,
+		MaxDesktopClientBacklog:  defaultMonitoringDailyMaxDesktopClientBacklog,
 	}
 	if len(options) > 0 {
 		in := options[0]
@@ -79,11 +85,17 @@ func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions
 		if in.MaxMaterializePerBrand > 0 {
 			out.MaxMaterializePerBrand = in.MaxMaterializePerBrand
 		}
+		if in.MaxDesktopDispatchPerRun > 0 {
+			out.MaxDesktopDispatchPerRun = in.MaxDesktopDispatchPerRun
+		}
+		if in.MaxDesktopClientBacklog > 0 {
+			out.MaxDesktopClientBacklog = in.MaxDesktopClientBacklog
+		}
 	}
-	if out.MaxMaterializePerPlan > out.MaxMaterializePerRun {
+	if out.MaxMaterializePerRun > 0 && out.MaxMaterializePerPlan > out.MaxMaterializePerRun {
 		out.MaxMaterializePerPlan = out.MaxMaterializePerRun
 	}
-	if out.MaxMaterializePerBrand > out.MaxMaterializePerPlan {
+	if out.MaxMaterializePerPlan > 0 && out.MaxMaterializePerBrand > out.MaxMaterializePerPlan {
 		out.MaxMaterializePerBrand = out.MaxMaterializePerPlan
 	}
 	return out
@@ -227,6 +239,7 @@ func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context, options ...M
 		stats["failed_plan_count"] = summary.FailedPlanCount
 		stats["failed_brand_count"] = summary.FailedBrandCount
 		stats["planned_task_count"] = summary.PlannedTaskCount
+		stats["eligible_task_count"] = summary.EligibleTaskCount
 		stats["due_task_count"] = summary.DueTaskCount
 		stats["created_task_count"] = summary.CreatedTaskCount
 		stats["desktop_task_count"] = summary.DesktopTaskCount
@@ -236,6 +249,8 @@ func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context, options ...M
 	stats["max_materialize_per_run"] = runOptions.MaxMaterializePerRun
 	stats["max_materialize_per_plan"] = runOptions.MaxMaterializePerPlan
 	stats["max_materialize_per_brand"] = runOptions.MaxMaterializePerBrand
+	stats["max_desktop_dispatch_per_run"] = runOptions.MaxDesktopDispatchPerRun
+	stats["max_desktop_client_backlog"] = runOptions.MaxDesktopClientBacklog
 	if err != nil {
 		if w.logger != nil {
 			w.logger.Warn("monitoring daily task generation failed", MonitoringWorkerErrorFields(err)...)
@@ -267,10 +282,10 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks(
 		TenantCount:  len(plans),
 	}
 	options := normalizeMonitoringDailyRunOptions(runOptions...)
-	globalRemaining := options.MaxMaterializePerRun
+	globalMaterializeRemaining := options.MaxMaterializePerRun
 
 	for _, plan := range plans {
-		if globalRemaining <= 0 {
+		if monitoringDailyLimitReached(globalMaterializeRemaining, options.MaxMaterializePerRun) {
 			break
 		}
 		executionOwner, ok, skipReason := s.resolveDailyMonitoringExecutionOwner(plan)
@@ -308,10 +323,15 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks(
 			continue
 		}
 
-		planRunRemaining := minMonitoringDailyInt(options.MaxMaterializePerPlan, globalRemaining)
+		planMaterializeRemaining := options.MaxMaterializePerPlan
+		if options.MaxMaterializePerRun > 0 && (planMaterializeRemaining <= 0 || globalMaterializeRemaining < planMaterializeRemaining) {
+			planMaterializeRemaining = globalMaterializeRemaining
+		}
+		planDesktopDispatchRemaining := monitoringDailyPlanDesktopDispatchBudget(options)
 
 		for _, brand := range brands {
-			if planRunRemaining <= 0 || globalRemaining <= 0 {
+			if monitoringDailyLimitReached(planMaterializeRemaining, options.MaxMaterializePerPlan) ||
+				monitoringDailyLimitReached(globalMaterializeRemaining, options.MaxMaterializePerRun) {
 				break
 			}
 
@@ -334,9 +354,14 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks(
 				continue
 			}
 
-			dispatchLimit := minMonitoringDailyInt(options.MaxMaterializePerBrand, planRunRemaining, globalRemaining)
+			materializeLimit := monitoringDailyBoundedLimit(
-			materializeLimit := dispatchLimit
+				len(candidates),
-			createdCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand(
+				options.MaxMaterializePerBrand,
+				planMaterializeRemaining,
+				globalMaterializeRemaining,
+			)
+			dispatchLimit := monitoringDailyDispatchLimit(len(candidates), planDesktopDispatchRemaining, planDesktopDispatchRemaining)
+			createdCount, eligibleCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand(
 				ctx,
 				projectionService,
 				plan,
@@ -348,6 +373,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks(
 				materializeLimit,
 				dispatchLimit,
 				executionOwner,
+				options.MaxDesktopClientBacklog,
 			)
 			if err != nil {
 				if abortErr := monitoringDailyAbortErr(ctx); abortErr != nil {
@@ -358,13 +384,18 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks(
 				continue
 			}
 
-			processedCount := int(maxMonitoringDailyInt(dueCount, desktopCount+deferredCount))
+			if options.MaxMaterializePerPlan > 0 {
-			if processedCount > 0 {
+				planMaterializeRemaining -= int(dueCount)
-				planRunRemaining -= processedCount
+			}
-				globalRemaining -= processedCount
+			if options.MaxMaterializePerRun > 0 {
+				globalMaterializeRemaining -= int(dueCount)
+			}
+			if planDesktopDispatchRemaining > 0 {
+				planDesktopDispatchRemaining -= int(desktopCount + deferredCount)
 			}
 			summary.BrandCount++
 			summary.PlannedTaskCount += int64(len(candidates))
+			summary.EligibleTaskCount += eligibleCount
 			summary.DueTaskCount += dueCount
 			summary.CreatedTaskCount += createdCount
 			summary.DesktopTaskCount += desktopCount
@@ -431,89 +462,105 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand(
 	materializeLimit int,
 	dispatchLimit int,
 	executionOwner string,
-) (int64, int64, int64, int64, error) {
+	maxDesktopClientBacklog int,
+) (int64, int64, int64, int64, int64, error) {
 	if materializeLimit <= 0 && dispatchLimit <= 0 {
-		return 0, 0, 0, 0, nil
+		return 0, 0, 0, 0, 0, nil
 	}
 	if executionOwner != "desktop_tasks" {
-		return 0, 0, 0, 0, nil
+		return 0, 0, 0, 0, 0, nil
 	}
 
 	var createdCount int64
+	var eligibleCount int64
 	var dueCount int64
-	if materializeLimit > 0 {
+	specs := make([]monitorDesktopTaskSpec, 0)
-		tx, err := s.monitoringPool.Begin(ctx)
-		if err != nil {
-			return 0, 0, 0, 0, response.ErrInternal(50041, "begin_failed", "failed to start monitoring daily task generation")
-		}
-		defer tx.Rollback(ctx)
-
-		locked, err := acquireMonitoringDailyBrandLock(ctx, tx, plan.TenantID, brand.BrandID, monitoringBusinessDateText(businessDay))
-		if err != nil {
-			return 0, 0, 0, 0, err
-		}
-
-		if locked {
-			existingKeys, err := s.loadDailyMonitoringTaskKeys(ctx, tx, plan.TenantID, brand.BrandID, businessDay)
-			if err != nil {
-				return 0, 0, 0, 0, err
-			}
-
-			dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, materializeLimit)
-			dueCount = int64(len(dueCandidates))
-			if len(dueCandidates) > 0 {
-				if err := s.syncMonitoringQuestionSnapshots(ctx, tx, plan.TenantID, brand.BrandID, questions, true); err != nil {
-					return 0, 0, 0, 0, err
-				}
-
-				createdCount, err = s.ensureDailyMonitoringCollectTasks(ctx, tx, plan, brand, businessDay, dueCandidates, executionOwner)
-				if err != nil {
-					return 0, 0, 0, 0, err
-				}
-
-				if projectionService != nil && createdCount > 0 {
-					if err := projectionService.rebuildBrandDailyProjection(ctx, tx, plan.TenantID, brand.BrandID, businessDay); err != nil {
-						return 0, 0, 0, 0, err
-					}
-				}
-			}
-		}
-
-		if err := tx.Commit(ctx); err != nil {
-			return 0, 0, 0, 0, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation")
-		}
-	}
 
 	if dispatchLimit <= 0 {
-		return createdCount, dueCount, 0, 0, nil
+		return createdCount, eligibleCount, dueCount, 0, 0, nil
 	}
-	dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates)
+	dispatchablePlatformIDs, err := s.loadDailyMonitorDispatchablePlatformIDs(ctx, plan.TenantID, plan.WorkspaceID, candidates, maxDesktopClientBacklog)
 	if err != nil {
-		return createdCount, dueCount, 0, 0, err
+		return createdCount, eligibleCount, dueCount, 0, 0, err
 	}
 	if len(dispatchablePlatformIDs) == 0 {
-		return createdCount, dueCount, 0, 0, nil
+		return createdCount, eligibleCount, dueCount, 0, 0, nil
 	}
-	specs, err := s.loadDueDailyMonitorDesktopTaskSpecs(
+	dispatchablePlatformSet := monitoringDailyPlatformSet(dispatchablePlatformIDs)
+	dispatchableCandidates := filterMonitoringDailyCandidatesByPlatform(candidates, dispatchablePlatformSet)
+	eligibleCount = int64(len(dispatchableCandidates))
+
+	retrySpecs, err := s.loadDueDailyMonitorDesktopTaskSpecs(
 		ctx,
 		s.monitoringPool,
 		plan.TenantID,
 		plan.WorkspaceID,
 		brand.BrandID,
 		businessDay,
-		now.Add(defaultMonitoringDailyLookahead),
+		now,
 		now.Add(-defaultMonitoringDailyDesktopRetryCooldown),
 		dispatchablePlatformIDs,
 		dispatchLimit,
 	)
 	if err != nil {
-		return createdCount, dueCount, 0, 0, err
+		return createdCount, eligibleCount, dueCount, 0, 0, err
 	}
-	if len(specs) == 0 {
+	if len(retrySpecs) > 0 {
-		return createdCount, dueCount, 0, 0, nil
+		specs = append(specs, retrySpecs...)
+		createdCount += int64(len(retrySpecs))
+		dueCount += int64(len(retrySpecs))
 	}
 
-	publishedTasks, deferredTaskIDs, err := s.dispatchMonitorDesktopTasks(ctx, specs)
+	remainingCreateLimit := dispatchLimit - len(specs)
+	if remainingCreateLimit <= 0 {
+		return s.dispatchDailyMonitoringDesktopSpecs(ctx, specs, maxDesktopClientBacklog, createdCount, eligibleCount, dueCount)
+	}
+
+	materializeLimit = monitoringDailyBoundedLimit(materializeLimit, dispatchLimit)
+	if materializeLimit > 0 && len(dispatchableCandidates) > 0 {
+		var materializeErr error
+		var newCreatedCount int64
+		var newDueCount int64
+		var newSpecs []monitorDesktopTaskSpec
+		newCreatedCount, newDueCount, newSpecs, materializeErr = s.createDispatchableDailyMonitoringTasksForBrand(
+			ctx,
+			projectionService,
+			plan,
+			brand,
+			businessDay,
+			now,
+			questions,
+			dispatchableCandidates,
+			monitoringDailyBoundedLimit(materializeLimit, remainingCreateLimit),
+			executionOwner,
+		)
+		if materializeErr != nil {
+			return createdCount, eligibleCount, dueCount, 0, 0, materializeErr
+		}
+		createdCount += newCreatedCount
+		dueCount += newDueCount
+		specs = append(specs, newSpecs...)
+	}
+	if len(specs) == 0 {
+		return createdCount, eligibleCount, dueCount, 0, 0, nil
+	}
+
+	return s.dispatchDailyMonitoringDesktopSpecs(ctx, specs, maxDesktopClientBacklog, createdCount, eligibleCount, dueCount)
+}
+
+func (s *MonitoringService) dispatchDailyMonitoringDesktopSpecs(
+	ctx context.Context,
+	specs []monitorDesktopTaskSpec,
+	maxDesktopClientBacklog int,
+	createdCount int64,
+	eligibleCount int64,
+	dueCount int64,
+) (int64, int64, int64, int64, int64, error) {
+	if len(specs) == 0 {
+		return createdCount, eligibleCount, dueCount, 0, 0, nil
+	}
+
+	publishedTasks, deferredTaskIDs, err := s.dispatchMonitorDesktopTasks(ctx, specs, maxDesktopClientBacklog)
 	if err != nil {
 		if s.logger != nil {
 			s.logger.Warn("daily monitor desktop dispatch failed; pending tasks will be retried",
@@ -521,7 +568,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand(
 				zap.Error(err),
 			)
 		}
-		return createdCount, dueCount, 0, 0, err
+		return createdCount, eligibleCount, dueCount, 0, 0, err
 	}
 
 	for _, task := range publishedTasks {
@@ -547,7 +594,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand(
 		cancel()
 	}
 
-	return createdCount, dueCount, int64(len(publishedTasks)), int64(len(deferredTaskIDs)), nil
+	return createdCount, eligibleCount, dueCount, int64(len(publishedTasks)), int64(len(deferredTaskIDs)), nil
 }
 
 func (s *MonitoringService) loadDailyMonitoringPlans(ctx context.Context) ([]monitoringDailyPlan, error) {
@@ -872,10 +919,72 @@ func (s *MonitoringService) loadDailyMonitoringTaskKeys(ctx context.Context, q m
 	return keys, nil
 }
 
+func (s *MonitoringService) createDispatchableDailyMonitoringTasksForBrand(
+	ctx context.Context,
+	projectionService *MonitoringCallbackService,
+	plan monitoringDailyPlan,
+	brand monitoringDailyBrand,
+	businessDay time.Time,
+	now time.Time,
+	questions []monitoringConfiguredQuestion,
+	candidates []monitoringDailyTaskCandidate,
+	limit int,
+	executionOwner string,
+) (int64, int64, []monitorDesktopTaskSpec, error) {
+	if limit <= 0 || len(candidates) == 0 {
+		return 0, 0, nil, nil
+	}
+
+	tx, err := s.monitoringPool.Begin(ctx)
+	if err != nil {
+		return 0, 0, nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring daily task generation")
+	}
+	defer tx.Rollback(ctx)
+
+	locked, err := acquireMonitoringDailyBrandLock(ctx, tx, plan.TenantID, brand.BrandID, monitoringBusinessDateText(businessDay))
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	if !locked {
+		return 0, 0, nil, nil
+	}
+
+	existingKeys, err := s.loadDailyMonitoringTaskKeys(ctx, tx, plan.TenantID, brand.BrandID, businessDay)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit)
+	dueCount := int64(len(dueCandidates))
+	if len(dueCandidates) == 0 {
+		if err := tx.Commit(ctx); err != nil {
+			return 0, dueCount, nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation")
+		}
+		return 0, dueCount, nil, nil
+	}
+
+	if err := s.syncMonitoringQuestionSnapshots(ctx, tx, plan.TenantID, brand.BrandID, questions, true); err != nil {
+		return 0, dueCount, nil, err
+	}
+	createdCount, specs, err := s.ensureDailyMonitoringCollectTasks(ctx, tx, plan, brand, businessDay, dueCandidates, executionOwner)
+	if err != nil {
+		return 0, dueCount, nil, err
+	}
+	if projectionService != nil && createdCount > 0 {
+		if err := projectionService.rebuildBrandDailyProjection(ctx, tx, plan.TenantID, brand.BrandID, businessDay); err != nil {
+			return 0, dueCount, nil, err
+		}
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return 0, dueCount, nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring daily task generation")
+	}
+	return createdCount, dueCount, specs, nil
+}
+
 func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs(
 	ctx context.Context,
 	tenantID, workspaceID int64,
 	candidates []monitoringDailyTaskCandidate,
+	maxDesktopClientBacklog int,
 ) ([]string, error) {
 	if len(candidates) == 0 {
 		return nil, nil
@@ -898,7 +1007,7 @@ func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs(
 		return nil, nil
 	}
 
-	targets, err := s.loadMonitorDesktopTaskTargets(ctx, tenantID, workspaceID, 0, specs)
+	targets, err := s.loadIdleMonitorDesktopTaskTargets(ctx, tenantID, workspaceID, 0, specs, maxDesktopClientBacklog)
 	if err != nil {
 		return nil, err
 	}
@@ -912,6 +1021,30 @@ func (s *MonitoringService) loadDailyMonitorDispatchablePlatformIDs(
 	return reconcileEnabledMonitoringPlatforms(platformIDs), nil
 }
 
+func monitoringDailyPlatformSet(platformIDs []string) map[string]struct{} {
+	result := make(map[string]struct{}, len(platformIDs))
+	for _, platformID := range platformIDs {
+		platformID = normalizeMonitoringPlatformID(platformID)
+		if platformID != "" {
+			result[platformID] = struct{}{}
+		}
+	}
+	return result
+}
+
+func filterMonitoringDailyCandidatesByPlatform(candidates []monitoringDailyTaskCandidate, platforms map[string]struct{}) []monitoringDailyTaskCandidate {
+	if len(candidates) == 0 || len(platforms) == 0 {
+		return nil
+	}
+	result := make([]monitoringDailyTaskCandidate, 0, len(candidates))
+	for _, candidate := range candidates {
+		if _, ok := platforms[normalizeMonitoringPlatformID(candidate.Platform.ID)]; ok {
+			result = append(result, candidate)
+		}
+	}
+	return result
+}
+
 func (s *MonitoringService) ensureDailyMonitoringCollectTasks(
 	ctx context.Context,
 	tx pgx.Tx,
@@ -920,12 +1053,12 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks(
 	businessDay time.Time,
 	candidates []monitoringDailyTaskCandidate,
 	executionOwner string,
-) (int64, error) {
+) (int64, []monitorDesktopTaskSpec, error) {
 	if len(candidates) == 0 {
-		return 0, nil
+		return 0, nil, nil
 	}
 	if strings.TrimSpace(executionOwner) != "desktop_tasks" {
-		return 0, response.ErrInternal(50041, "invalid_execution_owner", "daily monitoring tasks require desktop task execution")
+		return 0, nil, response.ErrInternal(50041, "invalid_execution_owner", "daily monitoring tasks require desktop task execution")
 	}
 
 	businessDate := monitoringBusinessDateText(businessDay)
@@ -940,7 +1073,7 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks(
 	for _, candidate := range candidates {
 		questionText, textErr := monitoringQuestionTextSnapshot(candidate.Question)
 		if textErr != nil {
-			return 0, textErr
+			return 0, nil, textErr
 		}
 		dispatchAfter := candidate.DispatchAfter
 		if dispatchAfter.IsZero() {
@@ -953,47 +1086,90 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks(
 		dispatchAfterValues = append(dispatchAfterValues, dispatchAfter)
 	}
 
-	tag, err := tx.Exec(ctx, `
+	rows, err := tx.Query(ctx, `
-		INSERT INTO monitoring_collect_tasks (
+		WITH input AS (
-			tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id,
+			SELECT *
-			collector_type, trigger_source, run_mode, business_date, planned_at, status,
+			FROM unnest(
-			dispatch_priority, dispatch_lane, target_client_id, dispatch_after,
+				$7::bigint[],
-			ingest_shard_key, execution_owner
+				$8::bytea[],
+				$9::text[],
+				$10::text[],
+				$11::timestamptz[]
+			) AS item(question_id, question_hash, question_text, ai_platform_id, dispatch_after)
+		),
+		upserted AS (
+			INSERT INTO monitoring_collect_tasks (
+				tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id,
+				collector_type, trigger_source, run_mode, business_date, planned_at, status,
+				dispatch_priority, dispatch_lane, target_client_id, dispatch_after,
+				ingest_shard_key, execution_owner
+			)
+			SELECT
+				$1,
+				$2,
+				input.question_id,
+				input.question_hash,
+				input.question_text,
+				input.ai_platform_id,
+				$3,
+				'automatic',
+				'desktop_standard',
+				$4::date,
+				input.dispatch_after,
+				'pending',
+				100,
+				'normal',
+				NULL,
+				input.dispatch_after,
+				$5,
+				$6
+			FROM input
+			ON CONFLICT (
+				tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
+			)
+			DO UPDATE SET
+				question_hash = EXCLUDED.question_hash,
+				question_text_snapshot = EXCLUDED.question_text_snapshot,
+				dispatch_after = LEAST(
+					COALESCE(monitoring_collect_tasks.dispatch_after, EXCLUDED.dispatch_after),
+					EXCLUDED.dispatch_after
+				),
+				updated_at = NOW()
+			WHERE monitoring_collect_tasks.status = 'pending'
+			  AND monitoring_collect_tasks.callback_received_at IS NULL
+			RETURNING *
+		),
+		claimed AS (
+			UPDATE monitoring_collect_tasks t
+			SET last_dispatched_at = NOW(),
+			    dispatch_attempts = COALESCE(t.dispatch_attempts, 0) + 1,
+			    updated_at = NOW()
+			FROM upserted u
+			WHERE t.id = u.id
+			  AND t.status = 'pending'
+			  AND t.callback_received_at IS NULL
+			  AND (t.dispatch_after IS NULL OR t.dispatch_after <= NOW())
+			  AND (
+			    t.last_dispatched_at IS NULL
+			    OR t.last_dispatched_at <= NOW() - $12::interval
+			  )
+			RETURNING t.*
 		)
 		SELECT
-			$1,
+			t.id,
-			$2,
+			t.tenant_id,
-			input.question_id,
+			t.ai_platform_id,
-			input.question_hash,
+			t.business_date,
-			input.question_text,
+			t.trigger_source,
-			input.ai_platform_id,
+			t.run_mode,
-			$3,
+			t.question_id,
-			'automatic',
+			t.question_hash,
-			'desktop_standard',
+			t.dispatch_priority,
-			$4::date,
+			t.dispatch_lane,
-			input.dispatch_after,
+			t.interrupt_generation,
-			'pending',
+			t.question_text_snapshot
-			100,
+		FROM claimed t
-			'normal',
+		ORDER BY t.dispatch_priority DESC, t.id ASC
-			NULL,
-			input.dispatch_after,
-			$5,
-			$6
-		FROM unnest(
-			$7::bigint[],
-			$8::bytea[],
-			$9::text[],
-			$10::text[],
-			$11::timestamptz[]
-		) AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after)
-		ON CONFLICT (
-			tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
-		)
-		DO UPDATE SET
-			question_hash = EXCLUDED.question_hash,
-			question_text_snapshot = EXCLUDED.question_text_snapshot,
-			updated_at = NOW()
-		WHERE btrim(COALESCE(monitoring_collect_tasks.question_text_snapshot, '')) = ''
 	`,
 		plan.TenantID,
 		brand.BrandID,
@@ -1006,12 +1182,50 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks(
 		questionTexts,
 		platformIDs,
 		dispatchAfterValues,
+		formatPgInterval(defaultMonitoringDailyDesktopRetryCooldown),
 	)
 	if err != nil {
-		return 0, response.ErrInternal(50041, "insert_failed", "failed to create daily monitoring tasks")
+		return 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create daily monitoring tasks")
 	}
+	defer rows.Close()
 
-	return tag.RowsAffected(), nil
+	specs := make([]monitorDesktopTaskSpec, 0, len(candidates))
+	for rows.Next() {
+		var (
+			spec         monitorDesktopTaskSpec
+			questionHash []byte
+			businessDay  time.Time
+		)
+		if scanErr := rows.Scan(
+			&spec.MonitorTaskID,
+			&spec.TenantID,
+			&spec.PlatformID,
+			&businessDay,
+			&spec.TriggerSource,
+			&spec.RunMode,
+			&spec.QuestionID,
+			&questionHash,
+			&spec.Priority,
+			&spec.Lane,
+			&spec.InterruptGeneration,
+			&spec.QuestionText,
+		); scanErr != nil {
+			return 0, nil, response.ErrInternal(50041, "scan_failed", "failed to parse created daily monitoring tasks")
+		}
+		spec.WorkspaceID = plan.WorkspaceID
+		spec.BusinessDate = businessDay.Format("2006-01-02")
+		spec.QuestionHash = encodeQuestionHash(questionHash)
+		spec.QuestionText = strings.TrimSpace(spec.QuestionText)
+		if spec.QuestionText == "" {
+			continue
+		}
+		spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText)
+		specs = append(specs, spec)
+	}
+	if err := rows.Err(); err != nil {
+		return 0, nil, response.ErrInternal(50041, "scan_failed", "failed to iterate created daily monitoring tasks")
+	}
+	return int64(len(specs)), specs, nil
 }
 
 func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs(
@@ -1178,52 +1392,65 @@ func buildMonitoringDailyTaskCandidates(
 	if len(candidates) > limit {
 		candidates = candidates[:limit]
 	}
-	assignMonitoringDailyScheduleTimes(tenantID, brandID, businessDay, candidates)
+	assignMonitoringDailyScheduleTimes(businessDay, candidates)
 	return candidates
 }
 
-func assignMonitoringDailyScheduleTimes(tenantID, brandID int64, businessDay time.Time, candidates []monitoringDailyTaskCandidate) {
+func assignMonitoringDailyScheduleTimes(businessDay time.Time, candidates []monitoringDailyTaskCandidate) {
 	if len(candidates) == 0 {
 		return
 	}
 
-	start := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset)
+	dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC()
-	window := defaultMonitoringDailyMaterializeWindow
-	if window <= 0 {
-		window = time.Hour
-	}
-	slot := window / time.Duration(len(candidates))
-	if slot <= 0 {
-		slot = time.Second
-	}
-	latest := start.Add(window)
-	businessDate := monitoringBusinessDateText(businessDay)
-
 	for i := range candidates {
-		jitter := time.Duration(monitoringDailyStableHash(
+		candidates[i].MaterializeAt = dueAt
-			"dispatch_jitter",
+		candidates[i].DispatchAfter = dueAt
-			int64Text(tenantID),
-			int64Text(brandID),
-			businessDate,
-			int64Text(candidates[i].Question.ID),
-			candidates[i].Platform.ID,
-		) % uint64(slot))
-		materializeAt := start.Add(time.Duration(i)*slot + jitter)
-		if materializeAt.After(latest) {
-			materializeAt = latest
-		}
-		candidates[i].MaterializeAt = materializeAt.UTC()
-		candidates[i].DispatchAfter = materializeAt.UTC()
 	}
 
 	sort.SliceStable(candidates, func(i, j int) bool {
-		if candidates[i].MaterializeAt.Equal(candidates[j].MaterializeAt) {
+		if candidates[i].sortKey == candidates[j].sortKey {
-			return candidates[i].sortKey < candidates[j].sortKey
+			if candidates[i].Question.ID == candidates[j].Question.ID {
+				return candidates[i].Platform.ID < candidates[j].Platform.ID
+			}
+			return candidates[i].Question.ID < candidates[j].Question.ID
 		}
-		return candidates[i].MaterializeAt.Before(candidates[j].MaterializeAt)
+		return candidates[i].sortKey < candidates[j].sortKey
 	})
 }
 
+func monitoringDailyLimitReached(remaining, configuredLimit int) bool {
+	return configuredLimit > 0 && remaining <= 0
+}
+
+func monitoringDailyBoundedLimit(limit int, caps ...int) int {
+	if limit <= 0 {
+		return 0
+	}
+	for _, cap := range caps {
+		if cap > 0 && cap < limit {
+			limit = cap
+		}
+	}
+	return limit
+}
+
+func monitoringDailyDispatchLimit(candidateCount, remaining, configuredLimit int) int {
+	if configuredLimit <= 0 {
+		return candidateCount
+	}
+	if remaining <= 0 {
+		return 0
+	}
+	return monitoringDailyBoundedLimit(candidateCount, remaining)
+}
+
+func monitoringDailyPlanDesktopDispatchBudget(options MonitoringDailyTaskRunOptions) int {
+	if options.MaxDesktopDispatchPerRun > 0 {
+		return options.MaxDesktopDispatchPerRun
+	}
+	return options.MaxDesktopClientBacklog
+}
+
 func selectDueMonitoringDailyCandidates(
 	candidates []monitoringDailyTaskCandidate,
 	existing map[monitoringDailyTaskKey]struct{},

server/internal/tenant/app/monitoring_daily_task_worker_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestBuildMonitoringDailyTaskCandidatesMaterializesAfterMidnight(t *testing.T) {
+func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *testing.T) {
 	businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation())
 	questions := []monitoringConfiguredQuestion{
 		{ID: 101, QuestionHash: []byte("q101")},
@@ -31,18 +31,96 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesAfterMidnight(t *testing.
 	candidates := buildMonitoringDailyTaskCandidates(7, 11, businessDay, questions, platforms, 4)
 
 	require.Len(t, candidates, 4)
-	start := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC()
+	dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC()
-	end := start.Add(defaultMonitoringDailyMaterializeWindow)
 	for i, candidate := range candidates {
-		assert.False(t, candidate.MaterializeAt.Before(start))
+		assert.Equal(t, dueAt, candidate.MaterializeAt)
-		assert.False(t, candidate.MaterializeAt.After(end))
 		assert.Equal(t, candidate.MaterializeAt, candidate.DispatchAfter)
 		if i > 0 {
-			assert.False(t, candidate.MaterializeAt.Before(candidates[i-1].MaterializeAt))
+			assert.LessOrEqual(t, candidates[i-1].sortKey, candidate.sortKey)
 		}
 	}
 }
 
+func TestNormalizeMonitoringDailyRunOptionsDefaultsDoNotThrottleByPlanOrBrand(t *testing.T) {
+	got := normalizeMonitoringDailyRunOptions()
+
+	assert.Equal(t, 0, got.MaxMaterializePerRun)
+	assert.Equal(t, 0, got.MaxMaterializePerPlan)
+	assert.Equal(t, 0, got.MaxMaterializePerBrand)
+	assert.Equal(t, 0, got.MaxDesktopDispatchPerRun)
+}
+
+func TestMonitoringDailyBoundedLimitTreatsZeroCapsAsUnlimited(t *testing.T) {
+	assert.Equal(t, 102, monitoringDailyBoundedLimit(102, 0, 0, 0))
+	assert.Equal(t, 64, monitoringDailyBoundedLimit(102, 0, 64, 0))
+	assert.Equal(t, 12, monitoringDailyBoundedLimit(102, 0, 64, 12))
+}
+
+func TestMonitoringDailyDispatchLimitStopsWhenRunCapacityIsExhausted(t *testing.T) {
+	assert.Equal(t, 102, monitoringDailyDispatchLimit(102, 0, 0))
+	assert.Equal(t, 12, monitoringDailyDispatchLimit(102, 12, 12))
+	assert.Equal(t, 0, monitoringDailyDispatchLimit(102, 0, 12))
+}
+
+func TestMonitoringDailyPlanDesktopDispatchBudgetUsesBacklogWhenUnconfigured(t *testing.T) {
+	assert.Equal(t, 12, monitoringDailyPlanDesktopDispatchBudget(MonitoringDailyTaskRunOptions{
+		MaxDesktopClientBacklog: 12,
+	}))
+	assert.Equal(t, 6, monitoringDailyPlanDesktopDispatchBudget(MonitoringDailyTaskRunOptions{
+		MaxDesktopDispatchPerRun: 6,
+		MaxDesktopClientBacklog:  12,
+	}))
+}
+
+func TestAutomaticDailySelectionIncludesAllUserQuestionsAndPlatforms(t *testing.T) {
+	businessDay := time.Date(2026, 6, 27, 0, 0, 0, 0, monitoringBusinessLocation())
+	questions := make([]monitoringConfiguredQuestion, 0, 17)
+	for id := int64(1); id <= 17; id++ {
+		questions = append(questions, monitoringConfiguredQuestion{
+			ID:           id,
+			QuestionText: "用户问题 " + int64Text(id),
+			QuestionHash: []byte("q" + int64Text(id)),
+		})
+	}
+	platforms := []monitoringPlatformMetadata{
+		{ID: "yuanbao"},
+		{ID: "kimi"},
+		{ID: "wenxin"},
+		{ID: "doubao"},
+		{ID: "deepseek"},
+		{ID: "qwen"},
+	}
+
+	candidates := buildMonitoringDailyTaskCandidates(
+		7,
+		7,
+		businessDay,
+		questions,
+		platforms,
+		len(questions)*len(platforms),
+	)
+	limit := monitoringDailyBoundedLimit(len(candidates), 0, 0, 0)
+	due := selectDueMonitoringDailyCandidates(
+		candidates,
+		nil,
+		monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset),
+		defaultMonitoringDailyLookahead,
+		limit,
+	)
+
+	require.Len(t, candidates, 17*6)
+	require.Len(t, due, 17*6)
+	seen := make(map[monitoringDailyTaskKey]struct{}, len(due))
+	for _, candidate := range due {
+		seen[monitoringDailyTaskKey{
+			QuestionID: candidate.Question.ID,
+			PlatformID: candidate.Platform.ID,
+		}] = struct{}{}
+	}
+	assert.Contains(t, seen, monitoringDailyTaskKey{QuestionID: 17, PlatformID: "qwen"})
+	assert.Contains(t, seen, monitoringDailyTaskKey{QuestionID: 17, PlatformID: "deepseek"})
+}
+
 func TestSelectDueMonitoringDailyCandidatesSkipsExistingAndHonorsLimit(t *testing.T) {
 	businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation())
 	candidates := buildMonitoringDailyTaskCandidates(
@@ -68,7 +146,7 @@ func TestSelectDueMonitoringDailyCandidatesSkipsExistingAndHonorsLimit(t *testin
 			PlatformID: candidates[0].Platform.ID,
 		}: {},
 	}
-	now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset + defaultMonitoringDailyMaterializeWindow)
+	now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset)
 
 	due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 2)
 
@@ -110,7 +188,7 @@ func TestSelectDueMonitoringDailyCandidatesContinuesPastExistingTasks(t *testing
 		}] = struct{}{}
 	}
 
-	now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset + defaultMonitoringDailyMaterializeWindow)
+	now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset)
 	due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 4)
 
 	require.Len(t, due, 2)
@@ -158,10 +236,10 @@ func TestEnsureDailyMonitoringCollectTasksPersistsQuestionTextSnapshot(t *testin
 	tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("INSERT 0 2")}
 	businessDay := time.Date(2026, 6, 22, 0, 0, 0, 0, time.UTC)
 
-	created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks(
+	created, specs, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks(
 		ctx,
 		tx,
-		monitoringDailyPlan{TenantID: 4},
+		monitoringDailyPlan{TenantID: 4, WorkspaceID: 8},
 		monitoringDailyBrand{BrandID: 8},
 		businessDay,
 		[]monitoringDailyTaskCandidate{
@@ -186,23 +264,27 @@ func TestEnsureDailyMonitoringCollectTasksPersistsQuestionTextSnapshot(t *testin
 	)
 
 	require.NoError(t, err)
-	assert.Equal(t, int64(2), created)
+	assert.Equal(t, int64(0), created)
-	assert.True(t, tx.execCalled)
+	assert.Empty(t, specs)
-	normalizedSQL := normalizeSQLWhitespace(tx.execSQL)
+	assert.True(t, tx.queryCalled)
+	normalizedSQL := normalizeSQLWhitespace(tx.querySQL)
 	assert.Contains(t, normalizedSQL, "question_text_snapshot, ai_platform_id")
 	assert.Contains(t, normalizedSQL, "$10::text[]")
-	assert.Contains(t, normalizedSQL, "AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after)")
+	assert.Contains(t, normalizedSQL, "AS item(question_id, question_hash, question_text, ai_platform_id, dispatch_after)")
 	assert.Contains(t, normalizedSQL, "question_text_snapshot = EXCLUDED.question_text_snapshot")
-	require.Len(t, tx.execArgs, 11)
+	assert.Contains(t, normalizedSQL, "WHERE monitoring_collect_tasks.status = 'pending'")
-	assert.Equal(t, []string{"幽灵门轨道长度定制", "幽灵门轨道长度定制"}, tx.execArgs[8])
+	assert.Contains(t, normalizedSQL, "t.dispatch_after IS NULL OR t.dispatch_after <= NOW()")
-	assert.Equal(t, []string{"qwen", "doubao"}, tx.execArgs[9])
+	assert.Contains(t, normalizedSQL, "SET last_dispatched_at = NOW()")
+	require.Len(t, tx.queryArgs, 12)
+	assert.Equal(t, []string{"幽灵门轨道长度定制", "幽灵门轨道长度定制"}, tx.queryArgs[8])
+	assert.Equal(t, []string{"qwen", "doubao"}, tx.queryArgs[9])
 }
 
 func TestEnsureDailyMonitoringCollectTasksRejectsEmptyQuestionText(t *testing.T) {
 	ctx := context.Background()
 	tx := &captureDailyMonitorTaskTx{}
 
-	created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks(
+	created, specs, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks(
 		ctx,
 		tx,
 		monitoringDailyPlan{TenantID: 4},
@@ -223,10 +305,26 @@ func TestEnsureDailyMonitoringCollectTasksRejectsEmptyQuestionText(t *testing.T)
 
 	require.Error(t, err)
 	assert.Equal(t, int64(0), created)
+	assert.Empty(t, specs)
 	assert.False(t, tx.execCalled)
+	assert.False(t, tx.queryCalled)
 	assert.Contains(t, err.Error(), "missing_question_text_snapshot")
 }
 
+func TestFilterMonitoringDailyCandidatesByPlatformKeepsOnlyOnlineDispatchablePlatforms(t *testing.T) {
+	candidates := []monitoringDailyTaskCandidate{
+		{Question: monitoringConfiguredQuestion{ID: 1}, Platform: monitoringPlatformMetadata{ID: "qwen"}},
+		{Question: monitoringConfiguredQuestion{ID: 1}, Platform: monitoringPlatformMetadata{ID: "doubao"}},
+		{Question: monitoringConfiguredQuestion{ID: 2}, Platform: monitoringPlatformMetadata{ID: "kimi"}},
+	}
+
+	filtered := filterMonitoringDailyCandidatesByPlatform(candidates, monitoringDailyPlatformSet([]string{"qwen", "kimi"}))
+
+	require.Len(t, filtered, 2)
+	assert.Equal(t, "qwen", filtered[0].Platform.ID)
+	assert.Equal(t, "kimi", filtered[1].Platform.ID)
+}
+
 func TestDecodeDailyMonitoringEnabledPlatformsReportsMalformedJSON(t *testing.T) {
 	platforms, err := decodeDailyMonitoringEnabledPlatforms([]byte(`["qwen"`))
 	require.Error(t, err)
@@ -332,6 +430,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) {
 		FailedPlanCount:   1,
 		FailedBrandCount:  2,
 		PlannedTaskCount:  20,
+		EligibleTaskCount: 12,
 		DueTaskCount:      8,
 		CreatedTaskCount:  6,
 		DesktopTaskCount:  4,
@@ -348,6 +447,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) {
 	assert.Equal(t, int64(2), snapshot.BrandFailureTotal)
 	assert.Equal(t, int64(1), snapshot.SkippedPlanTotal)
 	assert.Equal(t, int64(20), snapshot.PlannedTaskTotal)
+	assert.Equal(t, int64(12), snapshot.EligibleTaskTotal)
 	assert.Equal(t, int64(8), snapshot.DueTaskTotal)
 	assert.Equal(t, int64(6), snapshot.CreatedTaskTotal)
 	assert.Equal(t, int64(4), snapshot.DesktopTaskTotal)
@@ -361,6 +461,7 @@ func TestMonitoringDailyTaskMetricsRecordsSummaryCounters(t *testing.T) {
 	prometheus := MonitoringDailyTaskMetricsPrometheus()
 	assert.Contains(t, prometheus, `monitoring_daily_task_worker_runs_total{status="completed"} 1`)
 	assert.Contains(t, prometheus, `monitoring_daily_task_generation_failures_total{scope="plan"} 1`)
+	assert.Contains(t, prometheus, `monitoring_daily_task_generation_tasks_total{stage="eligible"} 12`)
 	assert.Contains(t, prometheus, `monitoring_daily_task_generation_tasks_total{stage="created"} 6`)
 	assert.Contains(t, prometheus, `monitoring_daily_task_worker_last_business_date_info{business_date="2026-04-24"} 1`)
 }
@@ -501,12 +602,15 @@ func (r emptyDailyMonitorRows) RawValues() [][]byte                          { r
 func (r emptyDailyMonitorRows) Conn() *pgx.Conn                              { return nil }
 
 type captureDailyMonitorTaskTx struct {
-	execCalled bool
+	execCalled  bool
-	execSQL    string
+	execSQL     string
-	execArgs   []any
+	execArgs    []any
-	execSQLs   []string
+	execSQLs    []string
-	execArgses [][]any
+	execArgses  [][]any
-	commandTag pgconn.CommandTag
+	queryCalled bool
+	querySQL    string
+	queryArgs   []any
+	commandTag  pgconn.CommandTag
 }
 
 func (tx *captureDailyMonitorTaskTx) Begin(context.Context) (pgx.Tx, error) { return tx, nil }
@@ -530,7 +634,10 @@ func (tx *captureDailyMonitorTaskTx) Exec(_ context.Context, sql string, argumen
 	tx.execArgses = append(tx.execArgses, arguments)
 	return tx.commandTag, nil
 }
-func (tx *captureDailyMonitorTaskTx) Query(context.Context, string, ...any) (pgx.Rows, error) {
+func (tx *captureDailyMonitorTaskTx) Query(_ context.Context, sql string, arguments ...any) (pgx.Rows, error) {
+	tx.queryCalled = true
+	tx.querySQL = sql
+	tx.queryArgs = arguments
 	return emptyDailyMonitorRows{}, nil
 }
 func (tx *captureDailyMonitorTaskTx) QueryRow(context.Context, string, ...any) pgx.Row {

server/internal/tenant/app/monitoring_metrics_prometheus.go

@@ -98,6 +98,7 @@ func (monitoringDailyTaskMetricsCollector) Collect(ch chan<- prometheus.Metric)
 	monitoringCounter(ch, "monitoring_daily_task_generation_failures_total", "Total non-fatal daily monitoring generation failures by scope.", float64(snapshot.BrandFailureTotal), []string{"scope"}, "brand")
 	monitoringCounter(ch, "monitoring_daily_task_generation_skipped_plans_total", "Total daily monitoring plans skipped before generation.", float64(snapshot.SkippedPlanTotal), nil)
 	monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.PlannedTaskTotal), []string{"stage"}, "planned")
+	monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.EligibleTaskTotal), []string{"stage"}, "eligible")
 	monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.DueTaskTotal), []string{"stage"}, "due")
 	monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.CreatedTaskTotal), []string{"stage"}, "created")
 	monitoringCounter(ch, "monitoring_daily_task_generation_tasks_total", "Total daily monitoring tasks seen by generation stage.", float64(snapshot.DesktopTaskTotal), []string{"stage"}, "desktop")

server/internal/tenant/app/monitoring_phase2_desktop_tasks.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -65,6 +66,11 @@ type monitorDesktopTaskTargetCandidate struct {
 	ClientOrder          int
 }
 
+type monitorDesktopTaskTargetOptions struct {
+	MaxClientBacklog   int
+	RequireIdleClient bool
+}
+
 func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 {
 	if len(specs) == 0 {
 		return nil
@@ -203,15 +209,34 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs(
 func (s *MonitoringService) dispatchMonitorDesktopTasks(
 	ctx context.Context,
 	specs []monitorDesktopTaskSpec,
+	maxClientBacklog ...int,
 ) ([]*repository.DesktopTask, []int64, error) {
 	if s == nil || s.businessPool == nil || len(specs) == 0 {
 		return nil, nil, nil
 	}
 
+	backlogLimit := 0
+	if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 {
+		backlogLimit = maxClientBacklog[0]
+	}
 	var targets map[string]monitorDesktopTaskTarget
 	if monitorDesktopTaskSpecsNeedTargetLookup(specs) {
 		var err error
-		targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs)
+		targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs, backlogLimit)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+	clientBacklog := make(map[uuid.UUID]int)
+	if backlogLimit > 0 {
+		var err error
+		clientBacklog, err = s.loadMonitorDesktopClientBacklog(
+			ctx,
+			specs[0].TenantID,
+			specs[0].WorkspaceID,
+			monitorDesktopTaskDispatchClientIDs(specs, targets),
+			backlogLimit,
+		)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -237,11 +262,22 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks(
 			spec.TargetClientID = target.ClientID
 		}
 
-		task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec)
+		task, shouldPublish, shouldDefer, upsertErr := s.upsertMonitorDesktopTask(
+			ctx,
+			tx,
+			repo,
+			spec,
+			func(oldClientID, newClientID uuid.UUID) bool {
+				return reserveMonitorDesktopClientBacklog(oldClientID, newClientID, clientBacklog, backlogLimit)
+			},
+			func(oldClientID, newClientID uuid.UUID) (bool, error) {
+				return s.reserveMonitorDesktopClientBacklogTx(ctx, tx, spec.TenantID, spec.WorkspaceID, oldClientID, newClientID, backlogLimit)
+			},
+		)
 		if upsertErr != nil {
 			return nil, nil, upsertErr
 		}
-		if fallbackLegacy {
+		if shouldDefer {
 			deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID)
 			continue
 		}
@@ -262,6 +298,26 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks(
 	return publishableTasks, deferredTaskIDs, nil
 }
 
+func monitorDesktopTaskDispatchClientIDs(specs []monitorDesktopTaskSpec, targets map[string]monitorDesktopTaskTarget) []uuid.UUID {
+	if len(specs) == 0 {
+		return nil
+	}
+	result := make([]uuid.UUID, 0, len(specs)*2)
+	for _, spec := range specs {
+		if spec.TargetClientID != uuid.Nil {
+			result = append(result, spec.TargetClientID)
+		}
+		if targets == nil {
+			continue
+		}
+		target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)]
+		if ok && target.ClientID != uuid.Nil {
+			result = append(result, target.ClientID)
+		}
+	}
+	return uniqueUUIDs(result)
+}
+
 func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) bool {
 	for _, spec := range specs {
 		if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil {
@@ -279,6 +335,34 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets(
 	ctx context.Context,
 	tenantID, workspaceID, userID int64,
 	specs []monitorDesktopTaskSpec,
+	maxClientBacklog ...int,
+) (map[string]monitorDesktopTaskTarget, error) {
+	backlogLimit := 0
+	if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 {
+		backlogLimit = maxClientBacklog[0]
+	}
+	return s.loadMonitorDesktopTaskTargetsWithOptions(ctx, tenantID, workspaceID, userID, specs, monitorDesktopTaskTargetOptions{
+		MaxClientBacklog: backlogLimit,
+	})
+}
+
+func (s *MonitoringService) loadIdleMonitorDesktopTaskTargets(
+	ctx context.Context,
+	tenantID, workspaceID, userID int64,
+	specs []monitorDesktopTaskSpec,
+	maxClientBacklog int,
+) (map[string]monitorDesktopTaskTarget, error) {
+	return s.loadMonitorDesktopTaskTargetsWithOptions(ctx, tenantID, workspaceID, userID, specs, monitorDesktopTaskTargetOptions{
+		MaxClientBacklog:   maxClientBacklog,
+		RequireIdleClient: true,
+	})
+}
+
+func (s *MonitoringService) loadMonitorDesktopTaskTargetsWithOptions(
+	ctx context.Context,
+	tenantID, workspaceID, userID int64,
+	specs []monitorDesktopTaskSpec,
+	options monitorDesktopTaskTargetOptions,
 ) (map[string]monitorDesktopTaskTarget, error) {
 	platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs)
 	if len(platformIDs) == 0 {
@@ -373,7 +457,15 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets(
 	if err != nil {
 		return nil, err
 	}
-	return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients), nil
+	backlogLookupLimit := options.MaxClientBacklog
+	if options.RequireIdleClient && backlogLookupLimit <= 0 {
+		backlogLookupLimit = 1
+	}
+	activeBacklog, err := s.loadMonitorDesktopClientBacklog(ctx, tenantID, workspaceID, clientIDs, backlogLookupLimit)
+	if err != nil {
+		return nil, err
+	}
+	return selectMonitorDesktopTaskTargetsWithOptions(candidates, accountPresence, onlineClients, activeBacklog, options), nil
 }
 
 func uniqueMonitorDesktopTaskPlatformIDs(specs []monitorDesktopTaskSpec) []string {
@@ -463,13 +555,70 @@ func (s *MonitoringService) loadOnlineMonitorDesktopClients(ctx context.Context,
 	return result, nil
 }
 
+func (s *MonitoringService) loadMonitorDesktopClientBacklog(
+	ctx context.Context,
+	tenantID, workspaceID int64,
+	clientIDs []uuid.UUID,
+	limit int,
+) (map[uuid.UUID]int, error) {
+	result := make(map[uuid.UUID]int)
+	clientIDs = uniqueUUIDs(clientIDs)
+	if s == nil || s.businessPool == nil || limit <= 0 || len(clientIDs) == 0 {
+		return result, nil
+	}
+
+	rows, err := s.businessPool.Query(ctx, `
+		SELECT target_client_id, COUNT(*)::int
+		FROM desktop_tasks
+		WHERE tenant_id = $1
+		  AND workspace_id = $2
+		  AND target_client_id = ANY($3::uuid[])
+		  AND kind = 'monitor'
+		  AND status IN ('queued', 'in_progress')
+		GROUP BY target_client_id
+	`, tenantID, workspaceID, clientIDs)
+	if err != nil {
+		return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to resolve monitor desktop client backlog")
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			clientID uuid.UUID
+			count    int
+		)
+		if scanErr := rows.Scan(&clientID, &count); scanErr != nil {
+			return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to parse monitor desktop client backlog")
+		}
+		result[clientID] = count
+	}
+	if err := rows.Err(); err != nil {
+		return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to iterate monitor desktop client backlog")
+	}
+	return result, nil
+}
+
 func selectMonitorDesktopTaskTargets(
 	candidates []monitorDesktopAccountCandidate,
 	accountPresence map[uuid.UUID][]desktopAccountPresenceClient,
 	onlineClients map[uuid.UUID]bool,
+	clientBacklog map[uuid.UUID]int,
+	maxClientBacklog int,
+) map[string]monitorDesktopTaskTarget {
+	return selectMonitorDesktopTaskTargetsWithOptions(candidates, accountPresence, onlineClients, clientBacklog, monitorDesktopTaskTargetOptions{
+		MaxClientBacklog: maxClientBacklog,
+	})
+}
+
+func selectMonitorDesktopTaskTargetsWithOptions(
+	candidates []monitorDesktopAccountCandidate,
+	accountPresence map[uuid.UUID][]desktopAccountPresenceClient,
+	onlineClients map[uuid.UUID]bool,
+	clientBacklog map[uuid.UUID]int,
+	options monitorDesktopTaskTargetOptions,
 ) map[string]monitorDesktopTaskTarget {
 	result := make(map[string]monitorDesktopTaskTarget)
-	bestByPlatform := make(map[string]monitorDesktopTaskTargetCandidate)
+	candidatesByPlatform := make(map[string][]monitorDesktopTaskTargetCandidate)
 	for _, candidate := range candidates {
 		platformID := normalizeMonitoringPlatformID(candidate.PlatformID)
 		if platformID == "" || candidate.AccountID == uuid.Nil {
@@ -495,9 +644,7 @@ func selectMonitorDesktopTaskTargets(
 					RecordOrder:          candidate.RecordOrder,
 					ClientOrder:          index,
 				}
-				if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) {
+				candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item)
-					bestByPlatform[platformID] = item
-				}
 			}
 		}
 		if candidate.DBClientID == nil || *candidate.DBClientID == uuid.Nil || !onlineClients[*candidate.DBClientID] {
@@ -516,16 +663,93 @@ func selectMonitorDesktopTaskTargets(
 			SourceRank:  1,
 			RecordOrder: candidate.RecordOrder,
 		}
-		if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) {
+		candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item)
-			bestByPlatform[platformID] = item
+	}
+
+	platformIDs := make([]string, 0, len(candidatesByPlatform))
+	for platformID := range candidatesByPlatform {
+		platformIDs = append(platformIDs, platformID)
+	}
+	sort.Strings(platformIDs)
+
+	for _, platformID := range platformIDs {
+		platformCandidates := candidatesByPlatform[platformID]
+		sort.SliceStable(platformCandidates, func(i, j int) bool {
+			return platformCandidates[i].betterThan(platformCandidates[j])
+		})
+		for _, candidate := range platformCandidates {
+			if monitorDesktopClientBacklogUnavailable(candidate.Target.ClientID, clientBacklog, options) {
+				continue
+			}
+			result[platformID] = candidate.Target
+			if options.MaxClientBacklog > 0 {
+				clientBacklog[candidate.Target.ClientID]++
+			}
+			break
 		}
 	}
-	for platformID, candidate := range bestByPlatform {
-		result[platformID] = candidate.Target
-	}
 	return result
 }
 
+func monitorDesktopClientBacklogUnavailable(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, options monitorDesktopTaskTargetOptions) bool {
+	if clientID == uuid.Nil {
+		return false
+	}
+	if options.RequireIdleClient && clientBacklog[clientID] > 0 {
+		return true
+	}
+	return monitorDesktopClientBacklogFull(clientID, clientBacklog, options.MaxClientBacklog)
+}
+
+func monitorDesktopClientBacklogFull(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool {
+	return maxClientBacklog > 0 && clientID != uuid.Nil && clientBacklog[clientID] >= maxClientBacklog
+}
+
+func reserveMonitorDesktopClientBacklog(oldClientID, newClientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool {
+	if maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID {
+		return true
+	}
+	if clientBacklog == nil {
+		return true
+	}
+	if monitorDesktopClientBacklogFull(newClientID, clientBacklog, maxClientBacklog) {
+		return false
+	}
+	clientBacklog[newClientID]++
+	if oldClientID != uuid.Nil && clientBacklog[oldClientID] > 0 {
+		clientBacklog[oldClientID]--
+	}
+	return true
+}
+
+func (s *MonitoringService) reserveMonitorDesktopClientBacklogTx(
+	ctx context.Context,
+	tx pgx.Tx,
+	tenantID, workspaceID int64,
+	oldClientID, newClientID uuid.UUID,
+	maxClientBacklog int,
+) (bool, error) {
+	if tx == nil || maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID {
+		return true, nil
+	}
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(newClientID)); err != nil {
+		return false, response.ErrInternal(50123, "desktop_client_backlog_lock_failed", "failed to lock monitor desktop client backlog")
+	}
+	var backlog int
+	if err := tx.QueryRow(ctx, `
+		SELECT COUNT(*)::int
+		FROM desktop_tasks
+		WHERE tenant_id = $1
+		  AND workspace_id = $2
+		  AND target_client_id = $3
+		  AND kind = 'monitor'
+		  AND status IN ('queued', 'in_progress')
+	`, tenantID, workspaceID, newClientID).Scan(&backlog); err != nil {
+		return false, response.ErrInternal(50124, "desktop_client_backlog_check_failed", "failed to check monitor desktop client backlog")
+	}
+	return backlog < maxClientBacklog, nil
+}
+
 func (candidate monitorDesktopTaskTargetCandidate) betterThan(current monitorDesktopTaskTargetCandidate) bool {
 	if candidate.HealthRank != current.HealthRank {
 		return candidate.HealthRank < current.HealthRank
@@ -559,14 +783,17 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 	tx pgx.Tx,
 	repo repository.DesktopTaskRepository,
 	spec monitorDesktopTaskSpec,
+	reserveClientBacklog func(oldClientID, newClientID uuid.UUID) bool,
+	reserveClientBacklogTx func(oldClientID, newClientID uuid.UUID) (bool, error),
 ) (*repository.DesktopTask, bool, bool, error) {
 	var (
 		existingDesktopID      uuid.UUID
+		existingTargetClientID uuid.UUID
 		existingStatus         string
 		existingLeaseExpiresAt pgtype.Timestamptz
 		existingActiveAttempt  pgtype.UUID
 	)
-	err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
+	err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingTargetClientID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
 	switch err {
 	case nil:
 		task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
@@ -583,6 +810,18 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 		if targetAccountID == uuid.Nil {
 			targetAccountID = task.TargetAccountID
 		}
+		if reserveClientBacklog != nil && !reserveClientBacklog(existingTargetClientID, spec.TargetClientID) {
+			return nil, false, true, nil
+		}
+		if reserveClientBacklogTx != nil {
+			allowed, reserveErr := reserveClientBacklogTx(existingTargetClientID, spec.TargetClientID)
+			if reserveErr != nil {
+				return nil, false, false, reserveErr
+			}
+			if !allowed {
+				return nil, false, true, nil
+			}
+		}
 		payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
 		if payloadErr != nil {
 			return nil, false, false, payloadErr
@@ -658,6 +897,18 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 	if targetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil {
 		return nil, false, true, nil
 	}
+	if reserveClientBacklog != nil && !reserveClientBacklog(uuid.Nil, spec.TargetClientID) {
+		return nil, false, true, nil
+	}
+	if reserveClientBacklogTx != nil {
+		allowed, reserveErr := reserveClientBacklogTx(uuid.Nil, spec.TargetClientID)
+		if reserveErr != nil {
+			return nil, false, false, reserveErr
+		}
+		if !allowed {
+			return nil, false, true, nil
+		}
+	}
 	payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
 	if payloadErr != nil {
 		return nil, false, false, payloadErr
@@ -718,7 +969,7 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 
 func upsertMonitorDesktopTaskActiveLookupSQL() string {
 	return `
-		SELECT desktop_id, status, lease_expires_at, active_attempt_id
+		SELECT desktop_id, target_client_id, status, lease_expires_at, active_attempt_id
 		FROM desktop_tasks
 		WHERE kind = 'monitor'
 		  AND monitor_task_id = $1

server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go

@@ -32,6 +32,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersPresentAccountClient(t *testing.T
 		map[uuid.UUID]bool{
 			liveClientID: true,
 		},
+		nil,
+		0,
 	)
 
 	assert.Equal(t, monitorDesktopTaskTarget{
@@ -56,6 +58,8 @@ func TestSelectMonitorDesktopTaskTargetsFallsBackToOnlineDBClient(t *testing.T)
 		map[uuid.UUID]bool{
 			clientID: true,
 		},
+		nil,
+		0,
 	)
 
 	assert.Equal(t, monitorDesktopTaskTarget{
@@ -78,6 +82,8 @@ func TestSelectMonitorDesktopTaskTargetsSkipsOfflineClients(t *testing.T) {
 		},
 		nil,
 		map[uuid.UUID]bool{},
+		nil,
+		0,
 	)
 
 	assert.Empty(t, targets)
@@ -111,6 +117,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersHealthierOnlineCandidate(t *testi
 			riskClientID: true,
 			liveClientID: true,
 		},
+		nil,
+		0,
 	)
 
 	assert.Equal(t, monitorDesktopTaskTarget{
@@ -144,6 +152,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount
 			earlierClientID: true,
 			laterClientID:   true,
 		},
+		nil,
+		0,
 	)
 
 	assert.Equal(t, monitorDesktopTaskTarget{
@@ -152,6 +162,147 @@ func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount
 	}, targets["doubao"])
 }
 
+func TestSelectMonitorDesktopTaskTargetsAppliesClientBackpressure(t *testing.T) {
+	fullClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701")
+	availableClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000702")
+	fullAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000801")
+	availableAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000802")
+
+	targets := selectMonitorDesktopTaskTargets(
+		[]monitorDesktopAccountCandidate{
+			{
+				PlatformID:  "qwen",
+				AccountID:   fullAccountID,
+				DBClientID:  &fullClientID,
+				HealthRank:  0,
+				RecordOrder: 0,
+			},
+			{
+				PlatformID:  "qwen",
+				AccountID:   availableAccountID,
+				DBClientID:  &availableClientID,
+				HealthRank:  1,
+				RecordOrder: 1,
+			},
+		},
+		nil,
+		map[uuid.UUID]bool{
+			fullClientID:      true,
+			availableClientID: true,
+		},
+		map[uuid.UUID]int{
+			fullClientID: 12,
+		},
+		12,
+	)
+
+	assert.Equal(t, monitorDesktopTaskTarget{
+		AccountID: availableAccountID,
+		ClientID:  availableClientID,
+	}, targets["qwen"])
+}
+
+func TestSelectMonitorDesktopTaskTargetsConsumesBackpressureWithinRun(t *testing.T) {
+	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000703")
+	qwenAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000803")
+	doubaoAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000804")
+
+	targets := selectMonitorDesktopTaskTargets(
+		[]monitorDesktopAccountCandidate{
+			{
+				PlatformID:  "doubao",
+				AccountID:   doubaoAccountID,
+				DBClientID:  &clientID,
+				HealthRank:  0,
+				RecordOrder: 0,
+			},
+			{
+				PlatformID:  "qwen",
+				AccountID:   qwenAccountID,
+				DBClientID:  &clientID,
+				HealthRank:  0,
+				RecordOrder: 1,
+			},
+		},
+		nil,
+		map[uuid.UUID]bool{
+			clientID: true,
+		},
+		map[uuid.UUID]int{
+			clientID: 11,
+		},
+		12,
+	)
+
+	assert.Len(t, targets, 1)
+	assert.Contains(t, targets, "doubao")
+	assert.NotContains(t, targets, "qwen")
+}
+
+func TestSelectMonitorDesktopTaskTargetsRequireIdleSkipsAnyActiveBacklog(t *testing.T) {
+	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000708")
+	accountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000808")
+
+	targets := selectMonitorDesktopTaskTargetsWithOptions(
+		[]monitorDesktopAccountCandidate{
+			{
+				PlatformID: "qwen",
+				AccountID:  accountID,
+				DBClientID: &clientID,
+			},
+		},
+		nil,
+		map[uuid.UUID]bool{
+			clientID: true,
+		},
+		map[uuid.UUID]int{
+			clientID: 1,
+		},
+		monitorDesktopTaskTargetOptions{
+			MaxClientBacklog:   12,
+			RequireIdleClient: true,
+		},
+	)
+
+	assert.Empty(t, targets)
+}
+
+func TestReserveMonitorDesktopClientBacklogAllowsExistingSameClientWithoutConsumingSlot(t *testing.T) {
+	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000704")
+	backlog := map[uuid.UUID]int{
+		clientID: 12,
+	}
+
+	ok := reserveMonitorDesktopClientBacklog(clientID, clientID, backlog, 12)
+
+	assert.True(t, ok)
+	assert.Equal(t, 12, backlog[clientID])
+}
+
+func TestReserveMonitorDesktopClientBacklogConsumesCapacityPerTask(t *testing.T) {
+	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000705")
+	backlog := map[uuid.UUID]int{
+		clientID: 11,
+	}
+
+	assert.True(t, reserveMonitorDesktopClientBacklog(uuid.Nil, clientID, backlog, 12))
+	assert.False(t, reserveMonitorDesktopClientBacklog(uuid.Nil, clientID, backlog, 12))
+	assert.Equal(t, 12, backlog[clientID])
+}
+
+func TestReserveMonitorDesktopClientBacklogMovesCapacityWhenRetargeting(t *testing.T) {
+	oldClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000706")
+	newClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000707")
+	backlog := map[uuid.UUID]int{
+		oldClientID: 4,
+		newClientID: 11,
+	}
+
+	assert.True(t, reserveMonitorDesktopClientBacklog(oldClientID, newClientID, backlog, 12))
+	assert.Equal(t, 3, backlog[oldClientID])
+	assert.Equal(t, 12, backlog[newClientID])
+}
+
 func TestMarshalMonitorDesktopTaskPayloadRequiresQuestionText(t *testing.T) {
 	payload, err := marshalMonitorDesktopTaskPayload(monitorDesktopTaskSpec{
 		PlatformID:   "qwen",

server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql

@@ -93,7 +93,7 @@ INSERT INTO ops.scheduler_jobs
     (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config)
 VALUES
     ('schedule_dispatch', '内容定时任务派发', 'content', '索引驱动的延迟队列派发，仅认领到期内容任务并投递生成队列。', true, 'interval', 15, 'Asia/Shanghai', 30, 100, '{}'::jsonb),
-    ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '按业务日小批量物化监控采集任务并派发到桌面任务队列。', true, 'interval', 300, 'Asia/Shanghai', 45, 64, '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb),
+    ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。', true, 'interval', 300, 'Asia/Shanghai', 45, 12, '{"max_desktop_client_backlog":12}'::jsonb),
     ('monitoring_result_recovery', '监控结果恢复', 'monitoring', '按回调结果队列索引恢复未成功入队的监控结果。', true, 'interval', 60, 'Asia/Shanghai', 30, 100, '{}'::jsonb),
     ('monitoring_lease_recovery', '监控租约回收', 'monitoring', '按过期租约索引限批回收超过租约时间仍未回调的监控任务。', true, 'interval', 1800, 'Asia/Shanghai', 30, 1000, '{}'::jsonb),
     ('monitoring_received_inspection', '监控 received 卡死检查', 'monitoring', '按 received watchdog 索引限批标记长时间未入队的监控任务。', true, 'interval', 300, 'Asia/Shanghai', 30, 100, '{"threshold":"15m"}'::jsonb),

server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql

@@ -11,10 +11,11 @@ SET description = '索引驱动的延迟队列派发，仅认领到期内容任
 WHERE job_key = 'schedule_dispatch';
 
 UPDATE ops.scheduler_jobs
-SET description = '按业务日小批量物化监控采集任务并派发到桌面任务队列。',
+SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。',
     timeout_seconds = 45,
-    batch_size = 64,
+    batch_size = 12,
-    config = config || '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb,
+    config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand')
+             || '{"max_desktop_client_backlog":12}'::jsonb,
     updated_at = NOW()
 WHERE job_key = 'monitoring_daily_task';
 

server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.down.sql

@@ -0,0 +1,7 @@
+UPDATE ops.scheduler_jobs
+SET description = '按业务日小批量物化监控采集任务并派发到桌面任务队列。',
+    batch_size = 64,
+    config = (config - 'max_desktop_client_backlog')
+             || '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';

server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql

@@ -0,0 +1,7 @@
+UPDATE ops.scheduler_jobs
+SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。',
+    batch_size = 12,
+    config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand')
+             || '{"max_desktop_client_backlog":12}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';

server/migrations_ops/20260627184500_document_online_only_monitoring_daily.down.sql

@@ -0,0 +1,6 @@
+UPDATE ops.scheduler_jobs
+SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。',
+    batch_size = 12,
+    config = config || '{"max_desktop_client_backlog":12}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';

server/migrations_ops/20260627184500_document_online_only_monitoring_daily.up.sql

@@ -0,0 +1,7 @@
+UPDATE ops.scheduler_jobs
+SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。',
+    batch_size = 12,
+    config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand')
+             || '{"max_desktop_client_backlog":12}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';

server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.down.sql

@@ -0,0 +1,7 @@
+UPDATE ops.scheduler_jobs
+SET description = '客户端在线时按业务日滚动创建并派发监控采集任务，离线不生成待采积压。',
+    batch_size = 12,
+    config = (config - 'max_desktop_dispatch_per_run' - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand')
+             || '{"max_desktop_client_backlog":12}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';

server/migrations_ops/20260627191000_remove_monitoring_daily_global_dispatch_cap.up.sql

@@ -0,0 +1,7 @@
+UPDATE ops.scheduler_jobs
+SET description = '客户端在线时按业务日补齐当天监控采集任务，离线不生成待采积压；不设置全局派发上限，客户端侧由 backlog 控制。',
+    batch_size = NULL,
+    config = (config - 'batch_size' - 'max_desktop_dispatch_per_run' - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand')
+             || '{"max_desktop_client_backlog":12}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_daily_task';