feat: implement batch processing for daily monitoring tasks across platforms

server/internal/tenant/app/monitoring_daily_task_worker.go

@@ -26,13 +26,14 @@ const (
 	defaultMonitoringDailyLookahead               = 10 * time.Minute
 	defaultMonitoringDailyDesktopRetryCooldown    = 15 * time.Minute
 	defaultMonitoringDailyPublishTimeout          = 5 * time.Second
+	defaultMonitoringDailyPlatformBatchSize       = 2
 	defaultMonitoringDailyMaxMaterializePerRun    = 0
 	defaultMonitoringDailyMaxMaterializePerPlan   = 0
 	defaultMonitoringDailyMaxMaterializePerBrand  = 0
 	defaultMonitoringDailyMaxDesktopDispatch      = 0
 	defaultMonitoringDailyMaxDesktopClientBacklog = 12
 	monitoringDailyPlatformActiveWindow           = 24 * time.Hour
-	monitoringDailyMaterializeStartOffset         = 5 * time.Minute
+	monitoringDailyMaterializeStartOffset         = 0
 )
 
 type MonitoringDailyTaskWorker struct {
@@ -953,7 +954,7 @@ func (s *MonitoringService) createDispatchableDailyMonitoringTasksForBrand(
 	if err != nil {
 		return 0, 0, nil, err
 	}
-	dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit)
+	dueCandidates := selectDueMonitoringDailyBatchCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit)
 	dueCount := int64(len(dueCandidates))
 	if len(dueCandidates) == 0 {
 		if err := tx.Commit(ctx); err != nil {
@@ -1457,29 +1458,167 @@ func selectDueMonitoringDailyCandidates(
 	now time.Time,
 	lookahead time.Duration,
 	limit int,
+) []monitoringDailyTaskCandidate {
+	return selectDueMonitoringDailyCandidatesWithPlatformBatch(candidates, existing, now, lookahead, limit, 0)
+}
+
+func selectDueMonitoringDailyBatchCandidates(
+	candidates []monitoringDailyTaskCandidate,
+	existing map[monitoringDailyTaskKey]struct{},
+	now time.Time,
+	lookahead time.Duration,
+	limit int,
+) []monitoringDailyTaskCandidate {
+	return selectDueMonitoringDailyCandidatesWithPlatformBatch(candidates, existing, now, lookahead, limit, defaultMonitoringDailyPlatformBatchSize)
+}
+
+func selectDueMonitoringDailyCandidatesWithPlatformBatch(
+	candidates []monitoringDailyTaskCandidate,
+	existing map[monitoringDailyTaskKey]struct{},
+	now time.Time,
+	lookahead time.Duration,
+	limit int,
+	platformBatchSize int,
 ) []monitoringDailyTaskCandidate {
 	if len(candidates) == 0 || limit <= 0 {
 		return nil
 	}
 
 	cutoff := now.Add(lookahead)
-	result := make([]monitoringDailyTaskCandidate, 0, limit)
+	due := make([]monitoringDailyTaskCandidate, 0, minMonitoringDailyInt(limit, len(candidates)))
+	dueByPlatform := make(map[string][]monitoringDailyTaskCandidate)
 	for _, candidate := range candidates {
 		if candidate.MaterializeAt.IsZero() || candidate.MaterializeAt.After(cutoff) {
 			continue
 		}
+		platformID := normalizeMonitoringPlatformID(candidate.Platform.ID)
 		key := monitoringDailyTaskKey{
 			QuestionID: candidate.Question.ID,
-			PlatformID: normalizeMonitoringPlatformID(candidate.Platform.ID),
+			PlatformID: platformID,
 		}
 		if _, ok := existing[key]; ok {
 			continue
 		}
-		result = append(result, candidate)
+		due = append(due, candidate)
+		if platformID != "" {
+			dueByPlatform[platformID] = append(dueByPlatform[platformID], candidate)
+		}
+	}
+	if platformBatchSize <= 0 || len(dueByPlatform) <= platformBatchSize {
+		return limitMonitoringDailyCandidates(due, limit)
+	}
+
+	platformIDs := make([]string, 0, len(dueByPlatform))
+	for platformID := range dueByPlatform {
+		platformIDs = append(platformIDs, platformID)
+	}
+	selectedPlatforms := selectMonitoringDailyBatchPlatformIDs(platformIDs, now, platformBatchSize)
+	return selectMonitoringDailyCandidatesByPlatformQuota(dueByPlatform, selectedPlatforms, limit)
+}
+
+func limitMonitoringDailyCandidates(candidates []monitoringDailyTaskCandidate, limit int) []monitoringDailyTaskCandidate {
+	if limit <= 0 || len(candidates) == 0 {
+		return nil
+	}
+	if len(candidates) <= limit {
+		return candidates
+	}
+	return candidates[:limit]
+}
+
+func selectMonitoringDailyBatchPlatformIDs(platformIDs []string, now time.Time, limit int) []string {
+	if len(platformIDs) == 0 || limit <= 0 {
+		return nil
+	}
+	seen := make(map[string]struct{}, len(platformIDs))
+	ordered := make([]string, 0, len(platformIDs))
+	for _, platformID := range platformIDs {
+		platformID = normalizeMonitoringPlatformID(platformID)
+		if platformID == "" {
+			continue
+		}
+		if _, ok := seen[platformID]; ok {
+			continue
+		}
+		seen[platformID] = struct{}{}
+		ordered = append(ordered, platformID)
+	}
+	if len(ordered) == 0 {
+		return nil
+	}
+
+	bucket := int64(0)
+	if defaultMonitoringDailyTaskInterval > 0 {
+		bucket = now.Unix() / int64(defaultMonitoringDailyTaskInterval/time.Second)
+	}
+	businessDate := monitoringBusinessDateText(now)
+	sort.SliceStable(ordered, func(i, j int) bool {
+		left := monitoringDailyStableHash("platform_batch_order", businessDate, int64Text(bucket), ordered[i])
+		right := monitoringDailyStableHash("platform_batch_order", businessDate, int64Text(bucket), ordered[j])
+		if left == right {
+			leftIndex := monitoringPlatformSortIndex(ordered[i])
+			rightIndex := monitoringPlatformSortIndex(ordered[j])
+			if leftIndex == rightIndex {
+				return ordered[i] < ordered[j]
+			}
+			return leftIndex < rightIndex
+		}
+		return left < right
+	})
+	if len(ordered) > limit {
+		ordered = ordered[:limit]
+	}
+	return ordered
+}
+
+func selectMonitoringDailyCandidatesByPlatformQuota(
+	dueByPlatform map[string][]monitoringDailyTaskCandidate,
+	platformIDs []string,
+	limit int,
+) []monitoringDailyTaskCandidate {
+	if len(dueByPlatform) == 0 || len(platformIDs) == 0 || limit <= 0 {
+		return nil
+	}
+	result := make([]monitoringDailyTaskCandidate, 0, limit)
+	usedByPlatform := make(map[string]int, len(platformIDs))
+	baseQuota := limit / len(platformIDs)
+	extraQuota := limit % len(platformIDs)
+
+	for index, platformID := range platformIDs {
+		platformID = normalizeMonitoringPlatformID(platformID)
+		items := dueByPlatform[platformID]
+		if len(items) == 0 {
+			continue
+		}
+		quota := baseQuota
+		if index < extraQuota {
+			quota++
+		}
+		take := minMonitoringDailyInt(quota, len(items))
+		result = append(result, items[:take]...)
+		usedByPlatform[platformID] = take
+	}
+
+	for len(result) < limit {
+		added := false
+		for _, platformID := range platformIDs {
+			platformID = normalizeMonitoringPlatformID(platformID)
+			items := dueByPlatform[platformID]
+			used := usedByPlatform[platformID]
+			if used >= len(items) {
+				continue
+			}
+			result = append(result, items[used])
+			usedByPlatform[platformID] = used + 1
+			added = true
 			if len(result) >= limit {
 				return result
 			}
 		}
+		if !added {
+			break
+		}
+	}
 	return result
 }
 

server/internal/tenant/app/monitoring_daily_task_worker_test.go

@@ -31,7 +31,7 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *te
 	candidates := buildMonitoringDailyTaskCandidates(7, 11, businessDay, questions, platforms, 4)
 
 	require.Len(t, candidates, 4)
-	dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC()
+	dueAt := monitoringCanonicalBusinessDay(businessDay).UTC()
 	for i, candidate := range candidates {
 		assert.Equal(t, dueAt, candidate.MaterializeAt)
 		assert.Equal(t, candidate.MaterializeAt, candidate.DispatchAfter)
@@ -41,6 +41,141 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *te
 	}
 }
 
+func TestSelectDueMonitoringDailyCandidatesSplitsBatchAcrossTwoPlatforms(t *testing.T) {
+	businessDay := time.Date(2026, 6, 28, 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)),
+		})
+	}
+	candidates := buildMonitoringDailyTaskCandidates(
+		7,
+		7,
+		businessDay,
+		questions,
+		[]monitoringPlatformMetadata{
+			{ID: "deepseek"},
+			{ID: "qwen"},
+			{ID: "doubao"},
+			{ID: "kimi"},
+		},
+		17*4,
+	)
+	existing := make(map[monitoringDailyTaskKey]struct{}, 17)
+	for _, candidate := range candidates {
+		if candidate.Platform.ID == "deepseek" {
+			existing[monitoringDailyTaskKey{
+				QuestionID: candidate.Question.ID,
+				PlatformID: candidate.Platform.ID,
+			}] = struct{}{}
+		}
+	}
+	now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour)
+
+	due := selectDueMonitoringDailyBatchCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 12)
+
+	require.Len(t, due, 12)
+	counts := make(map[string]int)
+	for _, candidate := range due {
+		assert.NotEqual(t, "deepseek", candidate.Platform.ID)
+		counts[candidate.Platform.ID]++
+	}
+	require.Len(t, counts, 2)
+	for _, count := range counts {
+		assert.Equal(t, 6, count)
+	}
+}
+
+func TestSelectDueMonitoringDailyCandidatesLimitsSmallRemainderToTwoPlatforms(t *testing.T) {
+	businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation())
+	questions := make([]monitoringConfiguredQuestion, 0, 3)
+	for id := int64(1); id <= 3; id++ {
+		questions = append(questions, monitoringConfiguredQuestion{
+			ID:           id,
+			QuestionText: "用户问题 " + int64Text(id),
+			QuestionHash: []byte("q" + int64Text(id)),
+		})
+	}
+	candidates := buildMonitoringDailyTaskCandidates(
+		7,
+		7,
+		businessDay,
+		questions,
+		[]monitoringPlatformMetadata{
+			{ID: "qwen"},
+			{ID: "doubao"},
+			{ID: "kimi"},
+		},
+		9,
+	)
+	now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour)
+
+	due := selectDueMonitoringDailyBatchCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 12)
+
+	require.Len(t, due, 6)
+	counts := make(map[string]int)
+	for _, candidate := range due {
+		counts[candidate.Platform.ID]++
+	}
+	require.Len(t, counts, 2)
+	for _, count := range counts {
+		assert.Equal(t, 3, count)
+	}
+}
+
+func TestSelectDueMonitoringDailyCandidatesAllowsSinglePlatformRemainder(t *testing.T) {
+	businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation())
+	questions := make([]monitoringConfiguredQuestion, 0, 9)
+	for id := int64(1); id <= 9; id++ {
+		questions = append(questions, monitoringConfiguredQuestion{
+			ID:           id,
+			QuestionText: "用户问题 " + int64Text(id),
+			QuestionHash: []byte("q" + int64Text(id)),
+		})
+	}
+	candidates := buildMonitoringDailyTaskCandidates(
+		7,
+		7,
+		businessDay,
+		questions,
+		[]monitoringPlatformMetadata{{ID: "qwen"}},
+		9,
+	)
+	now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour)
+
+	due := selectDueMonitoringDailyBatchCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 12)
+
+	require.Len(t, due, 9)
+	for _, candidate := range due {
+		assert.Equal(t, "qwen", candidate.Platform.ID)
+	}
+}
+
+func TestSelectDueMonitoringDailyCandidatesUsesMidnightStart(t *testing.T) {
+	businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation())
+	candidates := buildMonitoringDailyTaskCandidates(
+		7,
+		7,
+		businessDay,
+		[]monitoringConfiguredQuestion{{
+			ID:           1,
+			QuestionText: "用户问题",
+			QuestionHash: []byte("q1"),
+		}},
+		[]monitoringPlatformMetadata{{ID: "qwen"}},
+		1,
+	)
+	now := monitoringCanonicalBusinessDay(businessDay)
+
+	due := selectDueMonitoringDailyCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 1)
+
+	require.Len(t, due, 1)
+	assert.Equal(t, "qwen", due[0].Platform.ID)
+}
+
 func TestNormalizeMonitoringDailyRunOptionsDefaultsDoNotThrottleByPlanOrBrand(t *testing.T) {
 	got := normalizeMonitoringDailyRunOptions()
 

server/internal/tenant/app/monitoring_phase2_desktop_tasks.go

@@ -682,7 +682,7 @@ func selectMonitorDesktopTaskTargetsWithOptions(
 				continue
 			}
 			result[platformID] = candidate.Target
-			if options.MaxClientBacklog > 0 {
+			if options.MaxClientBacklog > 0 && !options.RequireIdleClient {
 				clientBacklog[candidate.Target.ClientID]++
 			}
 			break

server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go

@@ -267,6 +267,40 @@ func TestSelectMonitorDesktopTaskTargetsRequireIdleSkipsAnyActiveBacklog(t *test
 	assert.Empty(t, targets)
 }
 
+func TestSelectMonitorDesktopTaskTargetsRequireIdleKeepsMultiplePlatformsOnSameIdleClient(t *testing.T) {
+	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000709")
+	qwenAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000809")
+	doubaoAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000810")
+
+	targets := selectMonitorDesktopTaskTargetsWithOptions(
+		[]monitorDesktopAccountCandidate{
+			{
+				PlatformID: "doubao",
+				AccountID:  doubaoAccountID,
+				DBClientID: &clientID,
+			},
+			{
+				PlatformID: "qwen",
+				AccountID:  qwenAccountID,
+				DBClientID: &clientID,
+			},
+		},
+		nil,
+		map[uuid.UUID]bool{
+			clientID: true,
+		},
+		map[uuid.UUID]int{},
+		monitorDesktopTaskTargetOptions{
+			MaxClientBacklog:   12,
+			RequireIdleClient: true,
+		},
+	)
+
+	require.Len(t, targets, 2)
+	assert.Equal(t, clientID, targets["doubao"].ClientID)
+	assert.Equal(t, clientID, targets["qwen"].ClientID)
+}
+
 func TestReserveMonitorDesktopClientBacklogAllowsExistingSameClientWithoutConsumingSlot(t *testing.T) {
 	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000704")
 	backlog := map[uuid.UUID]int{