feat(monitoring): fail same-client queued tasks on desktop authorization failure

When a desktop monitoring task fails with a non-retryable authorization failure
(login expired, challenge required, risk control), all pending same-client/same-platform
tasks for the same business day are immediately bulk-failed as non-retryable, avoiding
wasted execution attempts. Adds preflight authorization checks before task execution,
enriches failure payloads with disposition metadata (code, category, retryable flags),
and triggers account health reports on auth state degradation.

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

server/internal/tenant/app/monitoring_callback_service.go

@@ -249,6 +249,7 @@ type monitoringCollectTask struct {
 	SkipReason         sql.NullString
 	TriggerSource      string
 	QuestionText       string
+	TargetClientID     sql.NullString
 }
 
 type monitoringDesktopExecutionLease struct {
@@ -343,6 +344,12 @@ type monitoringTaskResultQueueMetaPatch struct {
 
 type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch
 
+type monitoringTaskFailureDisposition struct {
+	NonRetryable bool
+	Category     string
+	Code         string
+}
+
 func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
 	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
 	if err != nil {
@@ -1356,6 +1363,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 			completed_at,
 			skip_reason,
 			trigger_source,
+			COALESCE(target_client_id::text, '') AS target_client_id,
 			COALESCE((
 				SELECT question_text_snapshot
 				FROM monitoring_question_config_snapshots
@@ -1389,6 +1397,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 		&item.CompletedAt,
 		&item.SkipReason,
 		&item.TriggerSource,
+		&item.TargetClientID,
 		&item.QuestionText,
 	); err != nil {
 		if err == pgx.ErrNoRows {
@@ -1422,6 +1431,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 			completed_at,
 			skip_reason,
 			trigger_source,
+			COALESCE(target_client_id::text, '') AS target_client_id,
 			COALESCE((
 				SELECT question_text_snapshot
 				FROM monitoring_question_config_snapshots
@@ -1454,6 +1464,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 		&item.CompletedAt,
 		&item.SkipReason,
 		&item.TriggerSource,
+		&item.TargetClientID,
 		&item.QuestionText,
 	); err != nil {
 		if err == pgx.ErrNoRows {
@@ -1732,28 +1743,23 @@ func (s *MonitoringCallbackService) markTaskResultReceived(ctx context.Context,
 		return response.ErrBadRequest(40041, "invalid_task_id", "monitoring task is required")
 	}
 
-	nextStatus := "received"
-	if task.ExecutionOwner == "desktop_tasks" {
-		nextStatus = "pending"
-	}
-
 	if _, err := tx.Exec(ctx, `
 		UPDATE monitoring_collect_tasks
-		SET status = $2,
-		    callback_received_at = $3,
-		    completed_at = NULL,
-		    skip_reason = NULL,
-		    error_message = NULL,
-		    request_payload_json = (
+			SET status = 'received',
+			    callback_received_at = $2,
+			    completed_at = NULL,
+			    skip_reason = NULL,
+			    error_message = NULL,
+			    request_payload_json = (
 				CASE
 					WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
 						THEN '{}'::jsonb
-					ELSE request_payload_json
-				END
-			) || $4::jsonb,
-		    updated_at = NOW()
-		WHERE id = $1
-	`, task.ID, nextStatus, receivedAt, requestPayloadJSON); err != nil {
+						ELSE request_payload_json
+					END
+				) || $3::jsonb,
+			    updated_at = NOW()
+			WHERE id = $1
+	`, task.ID, receivedAt, requestPayloadJSON); err != nil {
 		return response.ErrInternal(50041, "task_receive_failed", "failed to persist monitoring callback payload")
 	}
 	return nil
@@ -2126,6 +2132,11 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
 	if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
 		return nil, err
 	}
+	if runStatus == "failed" {
+		if _, err := s.failRemainingMonitoringTasksForAuthorizationFailure(ctx, tx, task, req, completedAt); err != nil {
+			return nil, err
+		}
+	}
 
 	if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
 		return nil, err
@@ -2187,12 +2198,23 @@ func buildDesktopMonitorTaskErrorValue(task *monitoringCollectTask, req Monitori
 	if message == "" {
 		message = "monitoring task failed"
 	}
+	disposition := monitoringTaskFailureDispositionFromRequest(req)
 
 	result := map[string]any{
 		"code":               "monitoring_task_failed",
 		"message":            message,
 		"monitoring_task_id": task.ID,
 	}
+	if disposition.Code != "" {
+		result["code"] = disposition.Code
+	}
+	if disposition.NonRetryable {
+		result["retryable"] = false
+		result["non_retryable"] = true
+	}
+	if disposition.Category != "" {
+		result["failure_category"] = disposition.Category
+	}
 	if task != nil && strings.TrimSpace(task.AIPlatformID) != "" {
 		result["platform"] = strings.TrimSpace(task.AIPlatformID)
 	}
@@ -2202,6 +2224,83 @@ func buildDesktopMonitorTaskErrorValue(task *monitoringCollectTask, req Monitori
 	return result
 }
 
+func monitoringTaskFailureDispositionFromRequest(req MonitoringTaskResultRequest) monitoringTaskFailureDisposition {
+	runtimeError, _ := req.RawResponseJSON["runtime_error"].(map[string]interface{})
+	if len(runtimeError) == 0 {
+		if nested, ok := req.RawResponseJSON["error"].(map[string]interface{}); ok {
+			runtimeError = nested
+		}
+	}
+
+	disposition := monitoringTaskFailureDisposition{
+		Category: monitoringFailureString(runtimeError["failure_category"]),
+		Code:     monitoringFailureString(runtimeError["code"]),
+	}
+	message := monitoringFailureString(runtimeError["message"])
+	if message == "" {
+		message = strings.TrimSpace(optionalStringValue(req.ErrorMessage))
+	}
+	if monitoringFailureBoolEquals(runtimeError["retryable"], false) ||
+		monitoringFailureBoolEquals(runtimeError["non_retryable"], true) ||
+		monitoringAuthorizationFailureCode(disposition.Code) ||
+		monitoringAuthorizationFailureMessage(message) {
+		disposition.NonRetryable = true
+	}
+	if disposition.NonRetryable && disposition.Category == "" {
+		disposition.Category = "ai_platform_authorization"
+	}
+	return disposition
+}
+
+func monitoringFailureString(value interface{}) string {
+	text, ok := value.(string)
+	if !ok {
+		return ""
+	}
+	return strings.TrimSpace(text)
+}
+
+func monitoringFailureBoolEquals(value interface{}, expected bool) bool {
+	boolean, ok := value.(bool)
+	if !ok {
+		return false
+	}
+	return boolean == expected
+}
+
+func monitoringAuthorizationFailureCode(code string) bool {
+	normalized := strings.ToLower(strings.TrimSpace(code))
+	if normalized == "" {
+		return false
+	}
+	if normalized == "desktop_account_auth_expired" ||
+		normalized == "desktop_account_challenge_required" ||
+		normalized == "desktop_account_risk_control" {
+		return true
+	}
+	return strings.HasSuffix(normalized, "_login_required") ||
+		strings.HasSuffix(normalized, "_login_expired") ||
+		strings.HasSuffix(normalized, "_not_logged_in") ||
+		strings.HasSuffix(normalized, "_challenge_required")
+}
+
+func monitoringAuthorizationFailureMessage(message string) bool {
+	normalized := strings.ToLower(strings.TrimSpace(message))
+	if normalized == "" {
+		return false
+	}
+	return strings.Contains(normalized, "login required") ||
+		strings.Contains(normalized, "login expired") ||
+		strings.Contains(normalized, "not logged in") ||
+		strings.Contains(normalized, "unauthorized") ||
+		strings.Contains(normalized, "登录态失效") ||
+		strings.Contains(normalized, "登录态已失效") ||
+		strings.Contains(normalized, "登录已失效") ||
+		strings.Contains(normalized, "请重新登录") ||
+		strings.Contains(normalized, "请先登录") ||
+		strings.Contains(normalized, "未登录")
+}
+
 func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
 	ctx context.Context,
 	task *monitoringCollectTask,
@@ -2291,38 +2390,22 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
 		return response.ErrInternal(50041, "desktop_task_reload_failed", "failed to reload finalized phase2 desktop task")
 	}
 
+	bulkFailure, err := bulkFailQueuedDesktopTasksForAuthorizationFailure(ctx, tx, finalizedTask, errorJSON)
+	if err != nil {
+		return err
+	}
+
 	if err := tx.Commit(ctx); err != nil {
 		return response.ErrInternal(50041, "desktop_task_complete_commit_failed", "failed to commit phase2 desktop task completion")
 	}
 
-	title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(
-		finalizedTask.Kind,
-		finalizedTask.Payload,
-	)
-	if publishErr := publishDesktopTaskEvent(ctx, s.rabbitMQ, DesktopTaskEvent{
-		Type:                "task_completed",
-		TaskID:              finalizedTask.DesktopID.String(),
-		JobID:               finalizedTask.JobID.String(),
-		WorkspaceID:         finalizedTask.WorkspaceID,
-		TargetAccountID:     finalizedTask.TargetAccountID.String(),
-		TargetClientID:      finalizedTask.TargetClientID.String(),
-		Platform:            finalizedTask.Platform,
-		Title:               title,
-		BusinessDate:        businessDate,
-		SchedulerGroupKey:   schedulerGroupKey,
-		QuestionText:        questionText,
-		Status:              finalizedTask.Status,
-		Kind:                finalizedTask.Kind,
-		Priority:            finalizedTask.Priority,
-		Lane:                finalizedTask.Lane,
-		InterruptGeneration: finalizedTask.InterruptGeneration,
-		UpdatedAt:           finalizedTask.UpdatedAt,
-	}); publishErr != nil && s.logger != nil {
-		s.logger.Warn("phase2 desktop task completion event publish failed",
-			zap.Int64("monitor_task_id", task.ID),
-			zap.String("desktop_task_id", finalizedTask.DesktopID.String()),
-			zap.Error(publishErr),
-		)
+	if err := failMonitoringCollectTasksForDesktopAuthorizationFailure(ctx, s.monitoringPool, bulkFailure.MonitorTaskIDs, bulkFailure.ErrorPayload); err != nil {
+		return err
+	}
+
+	publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, finalizedTask, "task_completed")
+	for _, failedTask := range bulkFailure.Tasks {
+		publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, failedTask, "task_completed")
 	}
 
 	return nil
@@ -2715,6 +2798,102 @@ func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context,
 	return nil
 }
 
+func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorizationFailure(
+	ctx context.Context,
+	tx pgx.Tx,
+	task *monitoringCollectTask,
+	req MonitoringTaskResultRequest,
+	completedAt time.Time,
+) (int64, error) {
+	if task == nil {
+		return 0, nil
+	}
+	targetClientID := monitoringAuthorizationFailureTargetClientID(task)
+	if targetClientID == "" {
+		return 0, nil
+	}
+
+	disposition := monitoringTaskFailureDispositionFromRequest(req)
+	if !disposition.NonRetryable || disposition.Category != "ai_platform_authorization" {
+		return 0, nil
+	}
+
+	runtimeError := buildDesktopMonitorTaskErrorValue(task, req)
+	runtimeError["blocked_by_authorization_failure_task_id"] = task.ID
+	runtimeError["blocked_by_authorization_failure_platform"] = task.AIPlatformID
+	requestPayloadJSON, err := json.Marshal(map[string]any{
+		"runtime_error":                    runtimeError,
+		"blocked_by_authorization_failure": true,
+	})
+	if err != nil {
+		return 0, response.ErrInternal(50041, "task_payload_encode_failed", "failed to encode monitoring authorization failure payload")
+	}
+
+	message := strings.TrimSpace(optionalStringValue(req.ErrorMessage))
+	if message == "" {
+		message = monitoringFailureString(runtimeError["message"])
+	}
+	if message == "" {
+		message = "登录态已失效，任务未执行。"
+	}
+	executionOwner := strings.TrimSpace(task.ExecutionOwner)
+	if executionOwner == "" {
+		executionOwner = "legacy"
+	}
+
+	tag, err := tx.Exec(ctx, `
+		UPDATE monitoring_collect_tasks
+		SET status = 'failed',
+		    lease_token_hash = NULL,
+		    leased_to_executor = NULL,
+		    leased_at = NULL,
+		    lease_expires_at = NULL,
+		    callback_received_at = NOW(),
+		    completed_at = $7,
+		    skip_reason = NULL,
+		    error_message = $8,
+		    request_payload_json = (
+				CASE
+					WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
+						THEN '{}'::jsonb
+					ELSE request_payload_json
+				END
+			) || $9::jsonb,
+		    updated_at = NOW()
+		WHERE tenant_id = $1
+		  AND collector_type = $2
+		  AND business_date = $3::date
+		  AND ai_platform_id = $4
+		  AND COALESCE(execution_owner, 'legacy') = $10
+		  AND id <> $5
+		  AND status IN ('pending', 'leased', 'expired')
+		  AND callback_received_at IS NULL
+		  AND (
+		    target_client_id::text = $6
+		    OR leased_to_executor = $6
+		  )
+	`, task.TenantID, task.CollectorType, task.BusinessDate.Format("2006-01-02"), task.AIPlatformID, task.ID, targetClientID, completedAt, message, requestPayloadJSON, executionOwner)
+	if err != nil {
+		return 0, response.ErrInternal(50041, "task_update_failed", "failed to fail remaining monitoring tasks after authorization failure")
+	}
+	return tag.RowsAffected(), nil
+}
+
+func monitoringAuthorizationFailureTargetClientID(task *monitoringCollectTask) string {
+	if task == nil {
+		return ""
+	}
+	if task.TargetClientID.Valid {
+		if trimmed := strings.TrimSpace(task.TargetClientID.String); trimmed != "" {
+			return trimmed
+		}
+	}
+	if task.LeasedToExecutor.Valid {
+		return strings.TrimSpace(task.LeasedToExecutor.String)
+	}
+	return ""
+}
+
 func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, task *monitoringCollectTask) (*int64, int, int, error) {
 	var runID sql.NullInt64
 	if err := s.monitoringPool.QueryRow(ctx, `
@@ -3263,7 +3442,7 @@ func monitoringTaskReadyForQueuedResult(task *monitoringCollectTask) bool {
 		return false
 	}
 	if task.ExecutionOwner == "desktop_tasks" {
-		return task.Status == "pending" && task.CallbackReceivedAt.Valid
+		return (task.Status == "received" || task.Status == "pending") && task.CallbackReceivedAt.Valid
 	}
 	return task.Status == "received"
 }