feat(schedule): support KOL subscription targets and random time windows

Extend schedule tasks to dispatch KOL subscription prompts in addition
to prompt rules, add daily/weekly scheduling with fixed or random time
windows, and track dispatch outcomes (last run, status, consecutive
failures).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

server/internal/scheduler/schedule_dispatch_worker.go

@@ -3,6 +3,9 @@ package scheduler
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -22,12 +25,14 @@ const (
 	defaultScheduleDispatchTimeout  = 30 * time.Second
 	defaultScheduleDispatchBatch    = 20
 	scheduleCacheCleanupTimeout     = 30 * time.Second
+	maxInvalidScheduleFailures      = 3
 )
 
 type ScheduleDispatchWorker struct {
 	pool              *pgxpool.Pool
 	rabbitMQ          *rabbitmq.Client
 	promptRuleService *tenantapp.PromptRuleGenerationService
+	kolService        *tenantapp.KolGenerationService
 	cache             sharedcache.Cache
 	logger            *zap.Logger
 	cfg               scheduleDispatchRuntimeConfig
@@ -35,23 +40,30 @@ type ScheduleDispatchWorker struct {
 }
 
 type dueScheduleTask struct {
-	ID                int64
-	TenantID          int64
-	WorkspaceID       int64
-	OperatorID        *int64
-	PromptRuleID      int64
-	BrandID           *int64
-	Name              string
-	CronExpr          string
-	AutoPublish       bool
-	PublishAccountIDs []string
-	CoverAssetURL     *string
-	CoverImageAssetID *int64
-	EnableWebSearch   bool
-	GenerateCount     int
-	StartAt           *time.Time
-	EndAt             *time.Time
-	ScheduledFor      time.Time
+	ID                   int64
+	TenantID             int64
+	WorkspaceID          int64
+	OperatorID           *int64
+	TargetType           string
+	PromptRuleID         *int64
+	SubscriptionPromptID *int64
+	BrandID              *int64
+	Name                 string
+	CronExpr             string
+	ScheduleDays         []string
+	ScheduleTimeMode     string
+	RandomWindowStart    string
+	RandomWindowEnd      string
+	GenerationInput      map[string]interface{}
+	AutoPublish          bool
+	PublishAccountIDs    []string
+	CoverAssetURL        *string
+	CoverImageAssetID    *int64
+	EnableWebSearch      bool
+	GenerateCount        int
+	StartAt              *time.Time
+	EndAt                *time.Time
+	ScheduledFor         time.Time
 }
 
 type scheduleTaskCacheUpdate struct {
@@ -85,6 +97,11 @@ func NewScheduleDispatchWorker(
 	}
 }
 
+func (w *ScheduleDispatchWorker) WithKolGenerationService(svc *tenantapp.KolGenerationService) *ScheduleDispatchWorker {
+	w.kolService = svc
+	return w
+}
+
 func (w *ScheduleDispatchWorker) WithConfigProvider(provider config.Provider) *ScheduleDispatchWorker {
 	w.configProvider = provider
 	return w
@@ -237,7 +254,8 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
 	defer tx.Rollback(ctx)
 
 	rows, err := tx.Query(ctx, `
-		SELECT id, tenant_id, cron_expr, start_at, end_at, status
+		SELECT id, tenant_id, cron_expr, schedule_days, schedule_time_mode,
+		       random_window_start::text, random_window_end::text, start_at, end_at, status
 		FROM schedule_tasks
 		WHERE deleted_at IS NULL
 		  AND status = 'enabled'
@@ -255,29 +273,35 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
 	updates := make([]struct {
 		cacheUpdate scheduleTaskCacheUpdate
 		nextRunAt   *time.Time
+		invalidErr  error
 	}, 0, runtimeCfg.BatchSize)
 	for rows.Next() {
 		var (
-			id       int64
-			tenantID int64
-			cronExpr string
-			status   string
-			startAt  pgtype.Timestamptz
-			endAt    pgtype.Timestamptz
+			id                int64
+			tenantID          int64
+			cronExpr          string
+			scheduleDaysJSON  []byte
+			scheduleTimeMode  string
+			randomWindowStart string
+			randomWindowEnd   string
+			status            string
+			startAt           pgtype.Timestamptz
+			endAt             pgtype.Timestamptz
 		)
-		if err := rows.Scan(&id, &tenantID, &cronExpr, &startAt, &endAt, &status); err != nil {
+		if err := rows.Scan(&id, &tenantID, &cronExpr, &scheduleDaysJSON, &scheduleTimeMode, &randomWindowStart, &randomWindowEnd, &startAt, &endAt, &status); err != nil {
 			rows.Close()
 			return 0, err
 		}
 
 		var nextRunAt *time.Time
+		var nextErr error
 		if status == "enabled" {
-			nextRunAt, err = sharedschedule.ComputeNextRun(cronExpr, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), now)
-			if err != nil && w.logger != nil {
-				w.logger.Warn("schedule dispatch skipped invalid cron while hydrating next_run_at",
+			nextRunAt, nextErr = tenantapp.ComputeScheduleTaskNextRun(status, id, cronExpr, tenantapp.DecodeScheduleDayKeys(scheduleDaysJSON), scheduleTimeMode, randomWindowStart, randomWindowEnd, timePtrFromTimestamp(startAt), timePtrFromTimestamp(endAt), now)
+			if nextErr != nil && w.logger != nil {
+				w.logger.Warn("schedule dispatch skipped invalid schedule while hydrating next_run_at",
 					zap.Int64("schedule_task_id", id),
 					zap.String("cron_expr", cronExpr),
-					zap.Error(err),
+					zap.Error(nextErr),
 				)
 			}
 		}
@@ -285,12 +309,14 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
 		updates = append(updates, struct {
 			cacheUpdate scheduleTaskCacheUpdate
 			nextRunAt   *time.Time
+			invalidErr  error
 		}{
 			cacheUpdate: scheduleTaskCacheUpdate{
 				id:       id,
 				tenantID: tenantID,
 			},
-			nextRunAt: nextRunAt,
+			nextRunAt:  nextRunAt,
+			invalidErr: nextErr,
 		})
 	}
 	if err := rows.Err(); err != nil {
@@ -300,6 +326,23 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, run
 	rows.Close()
 
 	for _, update := range updates {
+		if update.invalidErr != nil {
+			message := truncateDispatchError(update.invalidErr)
+			if _, err := tx.Exec(ctx, `
+				UPDATE schedule_tasks
+				SET next_run_at = NULL,
+				    last_dispatch_status = 'failed',
+				    last_dispatch_error = NULLIF($1, ''),
+				    consecutive_failures = consecutive_failures + 1,
+				    status = CASE WHEN consecutive_failures + 1 >= $2 THEN 'disabled' ELSE status END,
+				    updated_at = NOW()
+				WHERE id = $3
+			`, message, maxInvalidScheduleFailures, update.cacheUpdate.id); err != nil {
+				return 0, err
+			}
+			count++
+			continue
+		}
 		if _, err := tx.Exec(ctx, `
 			UPDATE schedule_tasks
 			SET next_run_at = $1,
@@ -330,7 +373,9 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 	defer tx.Rollback(ctx)
 
 	rows, err := tx.Query(ctx, `
-		SELECT id, tenant_id, workspace_id, operator_id, prompt_rule_id, brand_id, name, cron_expr,
+		SELECT id, tenant_id, workspace_id, operator_id, target_type, prompt_rule_id,
+		       subscription_prompt_id, brand_id, name, cron_expr, schedule_days, schedule_time_mode,
+		       random_window_start::text, random_window_end::text, generation_input_json,
 		       auto_publish, publish_account_ids, cover_asset_url, cover_image_asset_id,
 		       enable_web_search, generate_count, start_at, end_at, next_run_at
 		FROM schedule_tasks
@@ -351,12 +396,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 	updates := make([]struct {
 		cacheUpdate scheduleTaskCacheUpdate
 		nextRun     *time.Time
+		invalidErr  error
 	}, 0, runtimeCfg.BatchSize)
 	for rows.Next() {
 		var (
 			task                  dueScheduleTask
 			workspaceID           pgtype.Int8
 			operatorID            pgtype.Int8
+			promptRuleID          pgtype.Int8
+			subscriptionPromptID  pgtype.Int8
+			generationInputJSON   []byte
+			scheduleDaysJSON      []byte
 			publishAccountIDsJSON []byte
 			startAt               pgtype.Timestamptz
 			endAt                 pgtype.Timestamptz
@@ -367,10 +417,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 			&task.TenantID,
 			&workspaceID,
 			&operatorID,
-			&task.PromptRuleID,
+			&task.TargetType,
+			&promptRuleID,
+			&subscriptionPromptID,
 			&task.BrandID,
 			&task.Name,
 			&task.CronExpr,
+			&scheduleDaysJSON,
+			&task.ScheduleTimeMode,
+			&task.RandomWindowStart,
+			&task.RandomWindowEnd,
+			&generationInputJSON,
 			&task.AutoPublish,
 			&publishAccountIDsJSON,
 			&task.CoverAssetURL,
@@ -391,6 +448,13 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 			task.WorkspaceID = workspaceID.Int64
 		}
 		task.OperatorID = int64PtrFromInt8(operatorID)
+		task.PromptRuleID = int64PtrFromInt8(promptRuleID)
+		task.SubscriptionPromptID = int64PtrFromInt8(subscriptionPromptID)
+		task.ScheduleDays = tenantapp.DecodeScheduleDayKeys(scheduleDaysJSON)
+		task.GenerationInput = decodeDueScheduleGenerationInput(generationInputJSON)
+		if task.TargetType == "" {
+			task.TargetType = tenantapp.ScheduleTargetPromptRule
+		}
 		task.PublishAccountIDs = decodeDueSchedulePublishAccountIDs(publishAccountIDsJSON)
 		if task.GenerateCount <= 0 {
 			task.GenerateCount = 1
@@ -401,28 +465,29 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 		}
 		task.ScheduledFor = *scheduledFor
 
-		nextRun, nextErr := sharedschedule.ComputeNextRun(task.CronExpr, task.StartAt, task.EndAt, now.Add(time.Nanosecond))
+		nextRun, nextErr := tenantapp.ComputeScheduleTaskNextRun("enabled", task.ID, task.CronExpr, task.ScheduleDays, task.ScheduleTimeMode, task.RandomWindowStart, task.RandomWindowEnd, task.StartAt, task.EndAt, now.Add(time.Nanosecond))
 		if nextErr != nil && w.logger != nil {
-			w.logger.Warn("schedule dispatch encountered invalid cron while claiming due schedule",
+			w.logger.Warn("schedule dispatch encountered invalid schedule while claiming due schedule",
 				zap.Int64("schedule_task_id", task.ID),
 				zap.String("cron_expr", task.CronExpr),
 				zap.Error(nextErr),
 			)
-			nextRun = nil
 		}
 
 		updates = append(updates, struct {
 			cacheUpdate scheduleTaskCacheUpdate
 			nextRun     *time.Time
+			invalidErr  error
 		}{
 			cacheUpdate: scheduleTaskCacheUpdate{
 				id:       task.ID,
 				tenantID: task.TenantID,
 			},
-			nextRun: nextRun,
+			nextRun:    nextRun,
+			invalidErr: nextErr,
 		})
 
-		if sharedschedule.IsRunAllowed(task.ScheduledFor, task.StartAt, task.EndAt) {
+		if nextErr == nil && sharedschedule.IsRunAllowed(task.ScheduledFor, task.StartAt, task.EndAt) {
 			tasks = append(tasks, task)
 		}
 	}
@@ -433,6 +498,22 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC
 	rows.Close()
 
 	for _, update := range updates {
+		if update.invalidErr != nil {
+			message := truncateDispatchError(update.invalidErr)
+			if _, err := tx.Exec(ctx, `
+				UPDATE schedule_tasks
+				SET next_run_at = NULL,
+				    last_dispatch_status = 'failed',
+				    last_dispatch_error = NULLIF($1, ''),
+				    consecutive_failures = consecutive_failures + 1,
+				    status = CASE WHEN consecutive_failures + 1 >= $2 THEN 'disabled' ELSE status END,
+				    updated_at = NOW()
+				WHERE id = $3
+			`, message, maxInvalidScheduleFailures, update.cacheUpdate.id); err != nil {
+				return nil, err
+			}
+			continue
+		}
 		if _, err := tx.Exec(ctx, `
 			UPDATE schedule_tasks
 			SET next_run_at = $1,
@@ -460,31 +541,20 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
 		generateCount = 1
 	}
 
+	failures := make([]error, 0)
 	for idx := 0; idx < generateCount; idx++ {
 		ctx, cancel := w.stageContext(parent)
-		resp, err := w.promptRuleService.EnqueueScheduledGeneration(ctx, tenantapp.SchedulePromptRuleGenerationInput{
-			ScheduleTaskID:    task.ID,
-			OperatorID:        task.OperatorID,
-			TenantID:          task.TenantID,
-			PromptRuleID:      task.PromptRuleID,
-			BrandID:           task.BrandID,
-			Name:              task.Name,
-			WorkspaceID:       task.WorkspaceID,
-			AutoPublish:       task.AutoPublish,
-			PublishAccountIDs: task.PublishAccountIDs,
-			CoverAssetURL:     task.CoverAssetURL,
-			CoverImageAssetID: task.CoverImageAssetID,
-			EnableWebSearch:   task.EnableWebSearch,
-			GenerateCount:     generateCount,
-			ScheduledFor:      task.ScheduledFor,
-		})
+		resp, err := w.enqueueDueGeneration(ctx, task, generateCount)
 		cancel()
 		if err != nil {
+			failures = append(failures, err)
 			if w.logger != nil {
 				w.logger.Warn("schedule dispatch enqueue generation failed",
 					zap.Int64("schedule_task_id", task.ID),
 					zap.Int64("tenant_id", task.TenantID),
-					zap.Int64("prompt_rule_id", task.PromptRuleID),
+					zap.String("target_type", task.TargetType),
+					zap.Int64("prompt_rule_id", derefInt64(task.PromptRuleID)),
+					zap.Int64("subscription_prompt_id", derefInt64(task.SubscriptionPromptID)),
 					zap.Int("generate_index", idx+1),
 					zap.Int("generate_count", generateCount),
 					zap.Time("scheduled_for", task.ScheduledFor.UTC()),
@@ -498,7 +568,9 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
 			w.logger.Info("schedule dispatch enqueued generation",
 				zap.Int64("schedule_task_id", task.ID),
 				zap.Int64("tenant_id", task.TenantID),
-				zap.Int64("prompt_rule_id", task.PromptRuleID),
+				zap.String("target_type", task.TargetType),
+				zap.Int64("prompt_rule_id", derefInt64(task.PromptRuleID)),
+				zap.Int64("subscription_prompt_id", derefInt64(task.SubscriptionPromptID)),
 				zap.Int64("article_id", resp.ArticleID),
 				zap.Int64("generation_task_id", resp.TaskID),
 				zap.Int("generate_index", idx+1),
@@ -507,6 +579,141 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
 			)
 		}
 	}
+	w.recordDispatchResult(context.Background(), task, generateCount, failures)
+}
+
+func (w *ScheduleDispatchWorker) enqueueDueGeneration(ctx context.Context, task dueScheduleTask, generateCount int) (*tenantapp.GenerateFromRuleResponse, error) {
+	switch task.TargetType {
+	case "", tenantapp.ScheduleTargetPromptRule:
+		if task.PromptRuleID == nil || *task.PromptRuleID <= 0 {
+			return nil, errors.New("schedule prompt_rule target missing prompt_rule_id")
+		}
+		return w.promptRuleService.EnqueueScheduledGeneration(ctx, tenantapp.SchedulePromptRuleGenerationInput{
+			ScheduleTaskID:    task.ID,
+			OperatorID:        task.OperatorID,
+			TenantID:          task.TenantID,
+			PromptRuleID:      *task.PromptRuleID,
+			BrandID:           task.BrandID,
+			Name:              task.Name,
+			WorkspaceID:       task.WorkspaceID,
+			AutoPublish:       task.AutoPublish,
+			PublishAccountIDs: task.PublishAccountIDs,
+			CoverAssetURL:     task.CoverAssetURL,
+			CoverImageAssetID: task.CoverImageAssetID,
+			EnableWebSearch:   task.EnableWebSearch,
+			GenerateCount:     generateCount,
+			ScheduledFor:      task.ScheduledFor,
+		})
+	case tenantapp.ScheduleTargetKolSubscriptionPrompt:
+		if w.kolService == nil {
+			return nil, errors.New("kol generation service is not configured")
+		}
+		if task.SubscriptionPromptID == nil || *task.SubscriptionPromptID <= 0 {
+			return nil, errors.New("schedule kol target missing subscription_prompt_id")
+		}
+		variables, _ := task.GenerationInput["variables"].(map[string]interface{})
+		kolResp, err := w.kolService.EnqueueScheduledGeneration(ctx, tenantapp.ScheduleKolGenerationInput{
+			ScheduleTaskID:       task.ID,
+			OperatorID:           task.OperatorID,
+			TenantID:             task.TenantID,
+			WorkspaceID:          task.WorkspaceID,
+			BrandID:              task.BrandID,
+			SubscriptionPromptID: *task.SubscriptionPromptID,
+			Name:                 task.Name,
+			Variables:            variables,
+			EnableWebSearch:      scheduleGenerationInputBool(task.GenerationInput, "enable_web_search", task.EnableWebSearch),
+			KnowledgeGroupIDs:    scheduleGenerationInputInt64s(task.GenerationInput, "knowledge_group_ids"),
+			AutoPublish:          task.AutoPublish,
+			PublishAccountIDs:    task.PublishAccountIDs,
+			CoverAssetURL:        task.CoverAssetURL,
+			CoverImageAssetID:    task.CoverImageAssetID,
+			GenerateCount:        generateCount,
+			ScheduledFor:         task.ScheduledFor,
+		})
+		if err != nil {
+			return nil, err
+		}
+		return &tenantapp.GenerateFromRuleResponse{
+			ArticleID: kolResp.ArticleID,
+			TaskID:    kolResp.GenerationTaskID,
+		}, nil
+	default:
+		return nil, fmt.Errorf("unsupported schedule target type: %s", task.TargetType)
+	}
+}
+
+func (w *ScheduleDispatchWorker) recordDispatchResult(ctx context.Context, task dueScheduleTask, generateCount int, failures []error) {
+	if w == nil || w.pool == nil {
+		return
+	}
+	failedCount := len(failures)
+	result := aggregateDispatchResult(generateCount, failedCount)
+	if result.status == "success" {
+		_, _ = w.pool.Exec(ctx, `
+			UPDATE schedule_tasks
+			SET last_run_at = $1,
+			    last_dispatch_status = $2,
+			    last_dispatch_error = NULL,
+			    consecutive_failures = 0,
+			    updated_at = NOW()
+			WHERE id = $3 AND tenant_id = $4
+		`, task.ScheduledFor, result.status, task.ID, task.TenantID)
+		return
+	}
+	message := truncateDispatchErrors(failures)
+	_, _ = w.pool.Exec(ctx, `
+		UPDATE schedule_tasks
+		SET last_run_at = $1,
+		    last_dispatch_status = $2,
+		    last_dispatch_error = NULLIF($3, ''),
+		    consecutive_failures = CASE WHEN $4 THEN consecutive_failures + 1 ELSE 0 END,
+		    updated_at = NOW()
+		WHERE id = $5 AND tenant_id = $6
+	`, task.ScheduledFor, result.status, message, result.countAsFailure, task.ID, task.TenantID)
+}
+
+type dispatchAggregateResult struct {
+	status         string
+	countAsFailure bool
+}
+
+func aggregateDispatchResult(generateCount, failedCount int) dispatchAggregateResult {
+	if generateCount <= 0 {
+		generateCount = 1
+	}
+	if failedCount <= 0 {
+		return dispatchAggregateResult{status: "success"}
+	}
+	if failedCount >= generateCount {
+		return dispatchAggregateResult{status: "failed", countAsFailure: true}
+	}
+	return dispatchAggregateResult{status: "partial"}
+}
+
+func truncateDispatchError(err error) string {
+	if err == nil {
+		return ""
+	}
+	message := err.Error()
+	if len(message) > 1000 {
+		return message[:1000]
+	}
+	return message
+}
+
+func truncateDispatchErrors(errs []error) string {
+	parts := make([]string, 0, len(errs))
+	for _, err := range errs {
+		if err == nil {
+			continue
+		}
+		parts = append(parts, err.Error())
+	}
+	message := strings.Join(parts, "; ")
+	if len(message) > 1000 {
+		return message[:1000]
+	}
+	return message
 }
 
 func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []dueScheduleTask) {
@@ -662,6 +869,24 @@ func int64PtrFromInt8(value pgtype.Int8) *int64 {
 	return &result
 }
 
+func derefInt64(value *int64) int64 {
+	if value == nil {
+		return 0
+	}
+	return *value
+}
+
+func decodeDueScheduleGenerationInput(raw []byte) map[string]interface{} {
+	if len(raw) == 0 {
+		return map[string]interface{}{}
+	}
+	var payload map[string]interface{}
+	if err := json.Unmarshal(raw, &payload); err != nil || payload == nil {
+		return map[string]interface{}{}
+	}
+	return payload
+}
+
 func decodeDueSchedulePublishAccountIDs(raw []byte) []string {
 	if len(raw) == 0 {
 		return []string{}
@@ -684,3 +909,65 @@ func decodeDueSchedulePublishAccountIDs(raw []byte) []string {
 	}
 	return result
 }
+
+func scheduleGenerationInputBool(input map[string]interface{}, key string, fallback bool) bool {
+	if len(input) == 0 {
+		return fallback
+	}
+	raw, ok := input[key]
+	if !ok || raw == nil {
+		return fallback
+	}
+	switch typed := raw.(type) {
+	case bool:
+		return typed
+	case string:
+		return typed == "true" || typed == "1" || typed == "yes"
+	case float64:
+		return typed != 0
+	default:
+		return fallback
+	}
+}
+
+func scheduleGenerationInputInt64s(input map[string]interface{}, key string) []int64 {
+	if len(input) == 0 {
+		return nil
+	}
+	raw, ok := input[key]
+	if !ok || raw == nil {
+		return nil
+	}
+	var ids []int64
+	switch typed := raw.(type) {
+	case []int64:
+		ids = typed
+	case []interface{}:
+		ids = make([]int64, 0, len(typed))
+		for _, item := range typed {
+			switch value := item.(type) {
+			case int64:
+				ids = append(ids, value)
+			case int:
+				ids = append(ids, int64(value))
+			case float64:
+				ids = append(ids, int64(value))
+			}
+		}
+	default:
+		return nil
+	}
+	seen := make(map[int64]struct{}, len(ids))
+	result := make([]int64, 0, len(ids))
+	for _, id := range ids {
+		if id <= 0 {
+			continue
+		}
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		result = append(result, id)
+	}
+	return result
+}

server/internal/scheduler/schedule_dispatch_worker_test.go

@@ -0,0 +1,30 @@
+package scheduler
+
+import "testing"
+
+func TestAggregateDispatchResult(t *testing.T) {
+	tests := []struct {
+		name             string
+		generateCount    int
+		failedCount      int
+		wantStatus       string
+		wantFailureCount bool
+	}{
+		{name: "all success", generateCount: 3, failedCount: 0, wantStatus: "success"},
+		{name: "partial success", generateCount: 3, failedCount: 1, wantStatus: "partial"},
+		{name: "all failed", generateCount: 3, failedCount: 3, wantStatus: "failed", wantFailureCount: true},
+		{name: "zero generate count defaults to one", generateCount: 0, failedCount: 1, wantStatus: "failed", wantFailureCount: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := aggregateDispatchResult(tt.generateCount, tt.failedCount)
+			if got.status != tt.wantStatus {
+				t.Fatalf("status = %q, want %q", got.status, tt.wantStatus)
+			}
+			if got.countAsFailure != tt.wantFailureCount {
+				t.Fatalf("countAsFailure = %v, want %v", got.countAsFailure, tt.wantFailureCount)
+			}
+		})
+	}
+}