geo/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go

package app

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"go.uber.org/zap"

	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
	"github.com/geo-platform/tenant-api/internal/shared/response"
	"github.com/geo-platform/tenant-api/internal/shared/stream"
	"github.com/geo-platform/tenant-api/internal/tenant/repository"
)

type monitoringCollectTaskQueryer interface {
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

type monitorDesktopTaskSpec struct {
	MonitorTaskID       int64
	TenantID            int64
	WorkspaceID         int64
	TargetClientID      uuid.UUID
	PlatformID          string
	BusinessDate        string
	TriggerSource       string
	RunMode             string
	QuestionID          int64
	QuestionHash        string
	QuestionText        string
	SchedulerGroupKey   string
	Priority            int
	Lane                string
	InterruptGeneration int
}

type monitorDesktopInterruptTarget struct {
	TaskID              string
	MonitorTaskID       int64
	TargetClientID      string
	WorkspaceID         int64
	InterruptGeneration int
}

func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 {
	if len(specs) == 0 {
		return nil
	}
	result := make([]int64, 0, len(specs))
	seen := make(map[int64]struct{}, len(specs))
	for _, spec := range specs {
		if spec.MonitorTaskID <= 0 {
			continue
		}
		if _, exists := seen[spec.MonitorTaskID]; exists {
			continue
		}
		seen[spec.MonitorTaskID] = struct{}{}
		result = append(result, spec.MonitorTaskID)
	}
	return result
}

func (s *MonitoringService) shouldUseDesktopTasksExecution(tenantID int64) bool {
	if s == nil {
		return false
	}
	if s.desktopTasksRolloutPercentage <= 0 {
		return false
	}
	if s.desktopTasksRolloutPercentage >= 100 {
		return true
	}
	mod := tenantID % 100
	if mod < 0 {
		mod = -mod
	}
	return int(mod) < s.desktopTasksRolloutPercentage
}

func (s *MonitoringService) loadMonitorDesktopTaskSpecs(
	ctx context.Context,
	q monitoringCollectTaskQueryer,
	tenantID, workspaceID int64,
	businessDate time.Time,
	questionIDs []int64,
	platformIDs []string,
	targetClientID uuid.UUID,
) ([]monitorDesktopTaskSpec, error) {
	if q == nil || len(questionIDs) == 0 || len(platformIDs) == 0 {
		return nil, nil
	}

	rows, err := q.Query(ctx, `
		SELECT
			t.id,
			t.tenant_id,
			t.ai_platform_id,
			t.business_date,
			t.trigger_source,
			t.run_mode,
			t.question_id,
			t.question_hash,
			t.dispatch_priority,
			t.dispatch_lane,
			t.interrupt_generation,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots s
				WHERE s.tenant_id = t.tenant_id
				  AND s.brand_id = t.brand_id
				  AND s.question_id = t.question_id
				  AND s.question_hash = t.question_hash
				  AND s.deleted_at IS NULL
				ORDER BY s.projected_at DESC, s.id DESC
				LIMIT 1
			), '') AS question_text_snapshot
		FROM monitoring_collect_tasks t
		WHERE t.tenant_id = $1
		  AND t.collector_type = $2
		  AND t.business_date = $3::date
		  AND t.status = 'pending'
		  AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks'
		  AND t.target_client_id = $4
		  AND t.question_id = ANY($5)
		  AND t.ai_platform_id = ANY($6)
		ORDER BY t.dispatch_priority DESC, t.id ASC
	`, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), targetClientID, questionIDs, platformIDs)
	if err != nil {
		return nil, response.ErrInternal(50041, "query_failed", "failed to load phase2 monitor desktop task specs")
	}
	defer rows.Close()

	result := make([]monitorDesktopTaskSpec, 0)
	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 nil, response.ErrInternal(50041, "scan_failed", "failed to parse phase2 monitor desktop task specs")
		}
		spec.WorkspaceID = workspaceID
		spec.TargetClientID = targetClientID
		spec.BusinessDate = businessDay.Format("2006-01-02")
		spec.QuestionHash = encodeQuestionHash(questionHash)
		spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText)
		result = append(result, spec)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "scan_failed", "failed to iterate phase2 monitor desktop task specs")
	}
	return result, nil
}

func (s *MonitoringService) dispatchMonitorDesktopTasks(
	ctx context.Context,
	specs []monitorDesktopTaskSpec,
) ([]*repository.DesktopTask, []int64, error) {
	if s == nil || s.businessPool == nil || len(specs) == 0 {
		return nil, nil, nil
	}

	accountMap, err := s.loadMonitorDesktopAccountMap(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].TargetClientID)
	if err != nil {
		return nil, nil, err
	}

	tx, err := s.businessPool.Begin(ctx)
	if err != nil {
		return nil, nil, response.ErrInternal(50115, "desktop_task_begin_failed", "failed to start phase2 monitor desktop dispatch")
	}
	defer tx.Rollback(ctx)

	repo := repository.NewDesktopTaskRepository(tx)
	publishableTasks := make([]*repository.DesktopTask, 0, len(specs))
	fallbackLegacyTaskIDs := make([]int64, 0)
	for _, spec := range specs {
		task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec, accountMap)
		if upsertErr != nil {
			return nil, nil, upsertErr
		}
		if fallbackLegacy {
			fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID)
			continue
		}
		if shouldPublish && task != nil {
			publishableTasks = append(publishableTasks, task)
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, nil, response.ErrInternal(50116, "desktop_task_commit_failed", "failed to commit phase2 monitor desktop dispatch")
	}
	return publishableTasks, fallbackLegacyTaskIDs, nil
}

func (s *MonitoringService) loadMonitorDesktopAccountMap(
	ctx context.Context,
	tenantID, workspaceID int64,
	targetClientID uuid.UUID,
) (map[string]uuid.UUID, error) {
	rows, err := s.businessPool.Query(ctx, `
		SELECT DISTINCT ON (platform_id)
			platform_id,
			desktop_id,
			health
		FROM platform_accounts
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND client_id = $3
		  AND deleted_at IS NULL
		ORDER BY platform_id ASC,
			CASE health
				WHEN 'live' THEN 0
				WHEN 'risk' THEN 1
				WHEN 'captcha' THEN 2
				ELSE 3
			END ASC,
			verified_at DESC NULLS LAST,
			updated_at DESC,
			created_at DESC
	`, tenantID, workspaceID, targetClientID)
	if err != nil {
		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to resolve monitor desktop accounts")
	}
	defer rows.Close()

	result := make(map[string]uuid.UUID)
	for rows.Next() {
		var (
			platformID string
			desktopID  uuid.UUID
			health     string
		)
		if scanErr := rows.Scan(&platformID, &desktopID, &health); scanErr != nil {
			return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to parse monitor desktop accounts")
		}
		if strings.TrimSpace(health) != "live" {
			continue
		}
		result[platformID] = desktopID
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50117, "desktop_account_lookup_failed", "failed to iterate monitor desktop accounts")
	}
	return result, nil
}

func (s *MonitoringService) upsertMonitorDesktopTask(
	ctx context.Context,
	tx pgx.Tx,
	repo repository.DesktopTaskRepository,
	spec monitorDesktopTaskSpec,
	accountMap map[string]uuid.UUID,
) (*repository.DesktopTask, bool, bool, error) {
	var (
		existingDesktopID      uuid.UUID
		existingStatus         string
		existingLeaseExpiresAt pgtype.Timestamptz
		existingActiveAttempt  pgtype.UUID
	)
	err := tx.QueryRow(ctx, `
		SELECT desktop_id, status, lease_expires_at, active_attempt_id
		FROM desktop_tasks
		WHERE kind = 'monitor'
		  AND monitor_task_id = $1
		  AND status IN ('queued', 'in_progress', 'unknown')
		ORDER BY created_at DESC, id DESC
		LIMIT 1
		FOR UPDATE
	`, spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
	switch err {
	case nil:
		task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
		if getErr != nil {
			return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload monitor desktop task")
		}
		expiredInProgress := existingStatus == "in_progress" &&
			existingLeaseExpiresAt.Valid &&
			existingLeaseExpiresAt.Time.Before(time.Now().UTC())
		if existingStatus != "queued" && !expiredInProgress {
			return task, false, false, nil
		}
		targetAccountID := task.TargetAccountID
		if override, ok := accountMap[spec.PlatformID]; ok {
			targetAccountID = override
		}
		payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
		if payloadErr != nil {
			return nil, false, false, payloadErr
		}

		if expiredInProgress && existingActiveAttempt.Valid {
			expiredAttemptID, convErr := uuid.FromBytes(existingActiveAttempt.Bytes[:])
			if convErr == nil {
				expiredErrorJSON, marshalErr := marshalOptionalJSON(map[string]any{
					"reason":  "lease_expired_requeue",
					"message": "phase2 monitor desktop task lease expired before completion; task has been re-queued",
					"source":  "monitoring_collect_now",
				})
				if marshalErr != nil {
					return nil, false, false, response.ErrInternal(50119, "desktop_task_update_failed", "failed to encode expired phase2 monitor desktop attempt payload")
				}
				if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
					TaskID:      existingDesktopID,
					AttemptID:   expiredAttemptID,
					FinalStatus: literalStringPtr("aborted"),
					Error:       expiredErrorJSON,
				}); finishErr != nil && s.logger != nil {
					s.logger.Warn("phase2 expired monitor desktop attempt finish failed",
						zap.String("desktop_id", existingDesktopID.String()),
						zap.Int64("monitor_task_id", spec.MonitorTaskID),
						zap.Error(finishErr),
					)
				}
			}
		}

		if _, execErr := tx.Exec(ctx, `
			UPDATE desktop_tasks
			SET target_account_id = $2,
			    target_client_id = $3,
			    platform_id = $4,
			    payload = $5,
			    status = 'queued',
			    active_attempt_id = NULL,
			    lease_token_hash = NULL,
			    lease_expires_at = NULL,
			    priority = $6,
			    lane = $7,
			    lane_weight = $8,
			    source = $9,
			    scheduler_group_key = $10,
			    interrupt_generation = GREATEST(interrupt_generation, $11),
			    interrupted_at = CASE
			        WHEN $12 THEN COALESCE(interrupted_at, NOW())
			        ELSE interrupted_at
			    END,
			    interrupt_reason = CASE
			        WHEN $12 THEN 'lease_expired_requeue'
			        ELSE interrupt_reason
			    END,
			    enqueued_at = NOW(),
			    updated_at = NOW()
			WHERE desktop_id = $1
		`, existingDesktopID, targetAccountID, spec.TargetClientID, spec.PlatformID, payloadJSON, spec.Priority, spec.Lane, monitoringLaneWeight(spec.Lane), monitoringDesktopTaskSource(spec.TriggerSource), nullableString(optionalNonEmptyString(spec.SchedulerGroupKey)), spec.InterruptGeneration, expiredInProgress); execErr != nil {
			return nil, false, false, response.ErrInternal(50119, "desktop_task_update_failed", "failed to promote queued phase2 monitor desktop task")
		}
		task, getErr = repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
		if getErr != nil {
			return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload promoted monitor desktop task")
		}
		return task, true, false, nil
	case pgx.ErrNoRows:
	default:
		return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to inspect active phase2 monitor desktop task")
	}

	targetAccountID, ok := accountMap[spec.PlatformID]
	if !ok {
		return nil, false, true, nil
	}
	payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec)
	if payloadErr != nil {
		return nil, false, false, payloadErr
	}
	desktopID := uuid.New()
	if _, execErr := tx.Exec(ctx, `
		INSERT INTO desktop_tasks (
			desktop_id,
			job_id,
			tenant_id,
			workspace_id,
			target_account_id,
			target_client_id,
			platform_id,
			kind,
			payload,
			status,
			priority,
			lane,
			lane_weight,
			source,
			scheduler_group_key,
			monitor_task_id,
			control_flags,
			interrupt_generation,
			enqueued_at
		)
		VALUES (
			$1,
			$2,
			$3,
			$4,
			$5,
			$6,
			$7,
			'monitor',
			$8,
			'queued',
			$9,
			$10,
			$11,
			$12,
			$13,
			$14,
			'{}'::jsonb,
			$15,
			NOW()
		)
	`, desktopID, uuid.New(), spec.TenantID, spec.WorkspaceID, targetAccountID, spec.TargetClientID, spec.PlatformID, payloadJSON, spec.Priority, spec.Lane, monitoringLaneWeight(spec.Lane), monitoringDesktopTaskSource(spec.TriggerSource), nullableString(optionalNonEmptyString(spec.SchedulerGroupKey)), spec.MonitorTaskID, spec.InterruptGeneration); execErr != nil {
		return nil, false, false, response.ErrInternal(50120, "desktop_task_create_failed", "failed to create phase2 monitor desktop task")
	}
	task, getErr := repo.GetByDesktopID(ctx, desktopID, spec.WorkspaceID)
	if getErr != nil {
		return nil, false, false, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload created phase2 monitor desktop task")
	}
	return task, true, false, nil
}

func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Context, taskIDs []int64) error {
	if s == nil || s.monitoringPool == nil || len(taskIDs) == 0 {
		return nil
	}
	if _, err := s.monitoringPool.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET execution_owner = 'legacy',
		    updated_at = NOW()
		WHERE id = ANY($1)
	`, taskIDs); err != nil {
		return response.ErrInternal(50041, "update_failed", "failed to fallback phase2 monitor tasks to legacy execution")
	}
	return nil
}

func (s *MonitoringService) requestPhase2MonitorInterrupts(
	ctx context.Context,
	targetClientID uuid.UUID,
	excludedMonitorTaskIDs []int64,
	interruptGeneration int,
) ([]monitorDesktopInterruptTarget, error) {
	if s == nil || s.businessPool == nil {
		return nil, nil
	}

	queryArgs := []any{targetClientID, interruptGeneration}
	query := `
		UPDATE desktop_tasks
		SET control_flags = (
				CASE
					WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object'
						THEN '{}'::jsonb
					ELSE control_flags
				END
			) || jsonb_build_object(
				'interrupt_requested', true,
				'interrupt_generation', $2::integer,
				'interrupt_reason', 'collect_now_preempt'
			),
		    interrupt_generation = GREATEST(interrupt_generation, $2::integer),
		    interrupted_at = COALESCE(interrupted_at, NOW()),
		    interrupt_reason = 'collect_now_preempt',
		    updated_at = NOW()
		WHERE kind = 'monitor'
		  AND target_client_id = $1
		  AND status = 'in_progress'
		  AND lane IN ('normal', 'normal_boosted', 'retry')
		  AND monitor_task_id IS NOT NULL
	`
	if len(excludedMonitorTaskIDs) > 0 {
		query += ` AND NOT (monitor_task_id = ANY($3::bigint[]))`
		queryArgs = append(queryArgs, excludedMonitorTaskIDs)
	}
	query += `
		RETURNING desktop_id::text, monitor_task_id, target_client_id::text, workspace_id, interrupt_generation
	`

	rows, err := s.businessPool.Query(ctx, query, queryArgs...)
	if err != nil {
		if s.logger != nil {
			s.logger.Error("phase2 monitor interrupt query failed",
				zap.Error(err),
				zap.String("target_client_id", targetClientID.String()),
				zap.Int("interrupt_generation", interruptGeneration),
				zap.Int("excluded_monitor_task_count", len(excludedMonitorTaskIDs)),
			)
		}
		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to mark active phase2 monitor desktop tasks for interrupt")
	}
	defer rows.Close()

	result := make([]monitorDesktopInterruptTarget, 0)
	for rows.Next() {
		var item monitorDesktopInterruptTarget
		if scanErr := rows.Scan(&item.TaskID, &item.MonitorTaskID, &item.TargetClientID, &item.WorkspaceID, &item.InterruptGeneration); scanErr != nil {
			return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to parse active phase2 monitor interrupt targets")
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to iterate active phase2 monitor interrupt targets")
	}
	return result, nil
}

func (s *MonitoringService) deferQueuedPhase2MonitorTasks(
	ctx context.Context,
	targetClientID uuid.UUID,
	excludedMonitorTaskIDs []int64,
) ([]*repository.DesktopTask, error) {
	if s == nil || s.businessPool == nil {
		return nil, nil
	}

	queryArgs := []any{targetClientID}
	query := `
		UPDATE desktop_tasks
		SET priority = 50,
		    lane = 'retry',
		    lane_weight = 20,
		    source = 'retry_recovery',
		    control_flags = (
				CASE
					WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object'
						THEN '{}'::jsonb
					ELSE control_flags
				END
			) || jsonb_build_object(
				'deferred_by_collect_now', true,
				'deferred_at', NOW()::text
			),
		    updated_at = NOW()
		WHERE kind = 'monitor'
		  AND target_client_id = $1
		  AND status = 'queued'
		  AND lane IN ('normal', 'normal_boosted')
		  AND monitor_task_id IS NOT NULL
	`
	if len(excludedMonitorTaskIDs) > 0 {
		query += ` AND NOT (monitor_task_id = ANY($2::bigint[]))`
		queryArgs = append(queryArgs, excludedMonitorTaskIDs)
	}
	query += ` RETURNING desktop_id, workspace_id`

	rows, err := s.businessPool.Query(ctx, query, queryArgs...)
	if err != nil {
		return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to defer queued phase2 monitor desktop tasks")
	}
	defer rows.Close()

	repo := repository.NewDesktopTaskRepository(s.businessPool)
	result := make([]*repository.DesktopTask, 0)
	for rows.Next() {
		var (
			desktopID   uuid.UUID
			workspaceID int64
		)
		if scanErr := rows.Scan(&desktopID, &workspaceID); scanErr != nil {
			return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to parse deferred phase2 monitor desktop tasks")
		}
		task, getErr := repo.GetByDesktopID(ctx, desktopID, workspaceID)
		if getErr != nil {
			return nil, response.ErrInternal(50118, "desktop_task_lookup_failed", "failed to reload deferred phase2 monitor desktop task")
		}
		result = append(result, task)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50122, "desktop_task_defer_failed", "failed to iterate deferred phase2 monitor desktop tasks")
	}
	return result, nil
}

func publishMonitorDesktopTaskAvailable(
	ctx context.Context,
	clientID string,
	task *repository.DesktopTask,
	rabbitMQPublisher func(context.Context, DesktopTaskEvent) error,
	dispatchPublisher func(context.Context, stream.DesktopDispatchEvent) error,
) error {
	if task == nil || strings.TrimSpace(clientID) == "" {
		return nil
	}
	title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload)
	event := DesktopTaskEvent{
		Type:                "task_available",
		TaskID:              task.DesktopID.String(),
		JobID:               task.JobID.String(),
		WorkspaceID:         task.WorkspaceID,
		TargetAccountID:     task.TargetAccountID.String(),
		TargetClientID:      clientID,
		Platform:            task.Platform,
		Title:               title,
		BusinessDate:        businessDate,
		SchedulerGroupKey:   schedulerGroupKey,
		QuestionText:        questionText,
		Status:              task.Status,
		Kind:                task.Kind,
		Priority:            task.Priority,
		Lane:                task.Lane,
		InterruptGeneration: task.InterruptGeneration,
		UpdatedAt:           task.UpdatedAt,
	}
	if rabbitMQPublisher != nil {
		if err := rabbitMQPublisher(ctx, event); err != nil {
			return err
		}
	}
	if dispatchPublisher != nil {
		if err := dispatchPublisher(ctx, stream.DesktopDispatchEvent{
			Type:                event.Type,
			TaskID:              event.TaskID,
			JobID:               event.JobID,
			WorkspaceID:         event.WorkspaceID,
			TargetAccountID:     event.TargetAccountID,
			TargetClientID:      event.TargetClientID,
			Platform:            event.Platform,
			Title:               event.Title,
			BusinessDate:        event.BusinessDate,
			SchedulerGroupKey:   event.SchedulerGroupKey,
			QuestionText:        event.QuestionText,
			Status:              event.Status,
			Kind:                event.Kind,
			Priority:            event.Priority,
			Lane:                event.Lane,
			InterruptGeneration: event.InterruptGeneration,
			UpdatedAt:           event.UpdatedAt,
		}); err != nil {
			return err
		}
	}
	return nil
}

func publishPhase2MonitorControlEvent(
	ctx context.Context,
	client *rabbitmq.Client,
	logger *zap.Logger,
	target monitorDesktopInterruptTarget,
) {
	if client == nil || strings.TrimSpace(target.TargetClientID) == "" || strings.TrimSpace(target.TaskID) == "" {
		return
	}
	if err := publishDesktopDispatchEvent(ctx, client, logger, stream.DesktopDispatchEvent{
		Type:                "task_control",
		TaskID:              target.TaskID,
		WorkspaceID:         target.WorkspaceID,
		TargetClientID:      target.TargetClientID,
		Status:              "in_progress",
		Kind:                "monitor",
		Lane:                "high",
		InterruptGeneration: target.InterruptGeneration,
		Control:             "interrupt_requested",
		Reason:              "collect_now_preempt",
		UpdatedAt:           time.Now().UTC(),
	}); err != nil && logger != nil {
		logger.Warn("phase2 monitor control event publish failed",
			zap.String("task_id", target.TaskID),
			zap.Int64("monitor_task_id", target.MonitorTaskID),
			zap.Error(err),
		)
	}
}

func monitoringLaneWeight(lane string) int {
	switch strings.TrimSpace(lane) {
	case "high":
		return 70
	case "normal_boosted":
		return 50
	case "normal":
		return 40
	case "retry":
		return 20
	default:
		return 0
	}
}

func monitoringDesktopTaskSource(triggerSource string) string {
	if strings.EqualFold(strings.TrimSpace(triggerSource), "manual") {
		return "collect_now"
	}
	return "daily_schedule"
}

func monitoringSchedulerGroupKey(questionHash string, questionID int64, questionText string) string {
	if trimmed := strings.TrimSpace(questionHash); trimmed != "" {
		return trimmed
	}
	if questionID > 0 {
		return fmt.Sprintf("qid:%d", questionID)
	}
	if trimmed := strings.TrimSpace(questionText); trimmed != "" {
		return trimmed
	}
	return "monitor"
}

func marshalMonitorDesktopTaskPayload(spec monitorDesktopTaskSpec) ([]byte, error) {
	title := strings.TrimSpace(spec.QuestionText)
	platformName := platformDisplayName(spec.PlatformID)
	if title != "" {
		title = fmt.Sprintf("%s · %s", platformName, title)
	} else {
		title = fmt.Sprintf("监控任务 · %s", platformName)
	}
	payload, err := json.Marshal(map[string]any{
		"title":                  title,
		"business_date":          spec.BusinessDate,
		"scheduler_group_key":    spec.SchedulerGroupKey,
		"question_text":          spec.QuestionText,
		"question_id":            spec.QuestionID,
		"question_hash":          spec.QuestionHash,
		"ai_platform_id":         spec.PlatformID,
		"platform_name":          platformName,
		"collector_type":         monitoringCollectorType,
		"run_mode":               spec.RunMode,
		"trigger_source":         spec.TriggerSource,
		"monitoring_task_id":     spec.MonitorTaskID,
		"priority":               spec.Priority,
		"lane":                   spec.Lane,
		"interrupt_generation":   spec.InterruptGeneration,
		"legacy_monitoring_task": false,
	})
	if err != nil {
		return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode phase2 monitor desktop task payload")
	}
	return payload, nil
}

func optionalNonEmptyString(value string) *string {
	trimmed := strings.TrimSpace(value)
	if trimmed == "" {
		return nil
	}
	return &trimmed
}