feat(monitoring): scope authorization-failure cancellation to platform across accounts

When a monitor task's final account hits human verification or an
authorization failure, cancel the remaining queued desktop/monitoring
tasks for the whole platform instead of only the same account. Follow-up
tasks are now marked skipped/aborted (not failed) with a
platform_human_verification / platform_authorization_unavailable reason
and a user-facing cancellation message, and emit task_canceled lifecycle
events for monitor kinds.

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

server/internal/tenant/app/desktop_task_authorization_failure.go

@@ -36,6 +36,39 @@ type queuedDesktopAuthorizationFailureCandidate struct {
 	MonitorTaskID *int64
 }
 
+const (
+	monitoringPlatformHumanVerificationSkipReason = "platform_human_verification"
+	monitoringPlatformAuthorizationSkipReason     = "platform_authorization_unavailable"
+)
+
+func authorizationFailureFollowupCancellation(code, message string) (string, string) {
+	normalized := strings.ToLower(strings.TrimSpace(code + " " + message))
+	if strings.Contains(normalized, "challenge") ||
+		strings.Contains(normalized, "captcha") ||
+		strings.Contains(normalized, "risk_control") ||
+		strings.Contains(normalized, "人机验证") ||
+		strings.Contains(normalized, "人工验证") ||
+		strings.Contains(normalized, "验证码") ||
+		strings.Contains(normalized, "风控") {
+		return monitoringPlatformHumanVerificationSkipReason, "该模型没有其他可用账号，因触发人机验证，后续采集任务已自动取消。"
+	}
+	return monitoringPlatformAuthorizationSkipReason, "该模型当前没有可用账号，后续采集任务已自动取消。"
+}
+
+func desktopAuthorizationFollowupTaskStatus(kind string) string {
+	if strings.TrimSpace(kind) == "monitor" {
+		return "aborted"
+	}
+	return "failed"
+}
+
+func desktopAuthorizationFollowupEventType(task *repository.DesktopTask) string {
+	if task != nil && task.Kind == "monitor" && task.Status == "aborted" {
+		return "task_canceled"
+	}
+	return "task_completed"
+}
+
 func bulkFailQueuedDesktopTasksForAuthorizationFailure(
 	ctx context.Context,
 	tx pgx.Tx,
@@ -62,13 +95,27 @@ func bulkFailQueuedDesktopTasksForAuthorizationFailure(
 		payload := ensureDesktopAuthorizationFailurePayload(errorPayload, disposition, task, candidate.MonitorTaskID)
 		payload["blocked_by_authorization_failure_task_id"] = task.DesktopID.String()
 		payload["blocked_by_authorization_failure_platform"] = task.Platform
+		if task.Kind == "monitor" {
+			skipReason, cancellationMessage := authorizationFailureFollowupCancellation(disposition.Code, disposition.Message)
+			payload["source_authorization_failure_code"] = payload["code"]
+			payload["code"] = "desktop_monitor_platform_authorization_canceled"
+			payload["message"] = cancellationMessage
+			payload["canceled"] = true
+			payload["cancellation_reason"] = skipReason
+		}
 
 		candidateErrorJSON, marshalErr := json.Marshal(payload)
 		if marshalErr != nil {
 			return nil, response.ErrInternal(50122, "desktop_task_authorization_failure_payload_failed", "failed to encode non-retryable authorization failure")
 		}
 
-		updated, updateErr := failQueuedDesktopTaskForAuthorizationFailure(ctx, tx, candidate.DesktopID, candidateErrorJSON)
+		updated, updateErr := finishQueuedDesktopTaskForAuthorizationFailure(
+			ctx,
+			tx,
+			candidate.DesktopID,
+			desktopAuthorizationFollowupTaskStatus(task.Kind),
+			candidateErrorJSON,
+		)
 		if updateErr != nil {
 			return nil, updateErr
 		}
@@ -107,7 +154,7 @@ func loadQueuedDesktopAuthorizationFailureCandidates(
 		  AND kind = $3
 		  AND status = 'queued'
 		  AND target_client_id = $4
-		  AND target_account_id = $5
+		  AND (kind = 'monitor' OR target_account_id = $5)
 		  AND platform_id = $6
 		  AND desktop_id <> $7
 		ORDER BY COALESCE(enqueued_at, created_at) ASC, created_at ASC, desktop_id ASC
@@ -139,25 +186,28 @@ func loadQueuedDesktopAuthorizationFailureCandidates(
 	return candidates, nil
 }
 
-func failQueuedDesktopTaskForAuthorizationFailure(
+func finishQueuedDesktopTaskForAuthorizationFailure(
 	ctx context.Context,
 	tx pgx.Tx,
 	desktopID uuid.UUID,
+	status string,
 	errorJSON []byte,
 ) (*repository.DesktopTask, error) {
 	row := tx.QueryRow(ctx, `
 		UPDATE desktop_tasks AS t
-		SET status = 'failed',
+		SET status = $2,
 		    result = NULL,
-		    error = $2,
+		    error = $3,
 		    active_attempt_id = NULL,
 		    lease_token_hash = NULL,
 		    lease_expires_at = NULL,
+		    completed_at = NOW(),
 		    updated_at = NOW()
 		WHERE desktop_id = $1
 		  AND status = 'queued'
 		RETURNING `+desktopTaskRepositoryReturningColumns,
 		desktopID,
+		status,
 		errorJSON,
 	)
 	updated, err := scanRepositoryDesktopTask(row)
@@ -165,7 +215,7 @@ func failQueuedDesktopTaskForAuthorizationFailure(
 		if err == pgx.ErrNoRows {
 			return nil, nil
 		}
-		return nil, response.ErrInternal(50123, "desktop_task_authorization_failure_update_failed", "failed to fail queued desktop task after authorization failure")
+		return nil, response.ErrInternal(50123, "desktop_task_authorization_failure_update_failed", "failed to finish queued desktop task after authorization failure")
 	}
 	return updated, nil
 }
@@ -311,9 +361,14 @@ func failMonitoringCollectTasksForDesktopAuthorizationFailure(
 	}
 
 	errorMessage := desktopAuthorizationFailureString(errorPayload["message"])
-	if errorMessage == "" {
-		errorMessage = "登录态已失效，任务未执行。"
-	}
+	skipReason, cancellationMessage := authorizationFailureFollowupCancellation(
+		desktopAuthorizationFailureString(errorPayload["code"]),
+		errorMessage,
+	)
+	errorPayload["canceled"] = true
+	errorPayload["cancellation_reason"] = skipReason
+	errorPayload["source_authorization_failure_message"] = errorMessage
+	errorPayload["message"] = cancellationMessage
 	requestPayloadJSON, err := json.Marshal(map[string]any{
 		"runtime_error":                    errorPayload,
 		"blocked_by_authorization_failure": true,
@@ -324,28 +379,28 @@ func failMonitoringCollectTasksForDesktopAuthorizationFailure(
 
 	if _, err := pool.Exec(ctx, `
 		UPDATE monitoring_collect_tasks
-		SET status = 'failed',
+		SET status = 'skipped',
 		    lease_token_hash = NULL,
 		    leased_to_executor = NULL,
 		    leased_at = NULL,
 		    lease_expires_at = NULL,
 		    callback_received_at = NOW(),
 		    completed_at = NOW(),
-		    skip_reason = NULL,
-		    error_message = $2,
+		    skip_reason = $2,
+		    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
-			) || $3::jsonb,
+			) || $4::jsonb,
 		    updated_at = NOW()
 		WHERE id = ANY($1::bigint[])
 		  AND callback_received_at IS NULL
 		  AND status IN ('pending', 'leased', 'expired')
-	`, monitorTaskIDs, errorMessage, requestPayloadJSON); err != nil {
-		return response.ErrInternal(50125, "monitoring_authorization_failure_update_failed", "failed to fail monitoring tasks after authorization failure")
+	`, monitorTaskIDs, skipReason, cancellationMessage, requestPayloadJSON); err != nil {
+		return response.ErrInternal(50125, "monitoring_authorization_failure_update_failed", "failed to skip monitoring tasks after authorization failure")
 	}
 	return nil
 }

server/internal/tenant/app/desktop_task_service.go

@@ -1269,7 +1269,7 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De
 
 	s.publishTaskEvent(ctx, task, "task_completed")
 	for _, failedTask := range bulkFailure.Tasks {
-		s.publishTaskEvent(ctx, failedTask, "task_completed")
+		s.publishTaskEvent(ctx, failedTask, desktopAuthorizationFollowupEventType(failedTask))
 	}
 
 	view := buildDesktopTaskView(task)

server/internal/tenant/app/monitoring_callback_service.go

@@ -2683,7 +2683,7 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
 
 	publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, finalizedTask, "task_completed")
 	for _, failedTask := range bulkFailure.Tasks {
-		publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, failedTask, "task_completed")
+		publishDesktopTaskLifecycleEvent(ctx, s.rabbitMQ, s.logger, failedTask, desktopAuthorizationFollowupEventType(failedTask))
 	}
 
 	return nil
@@ -3260,13 +3260,6 @@ func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorization
 	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 == "" {
@@ -3275,6 +3268,18 @@ func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorization
 	if message == "" {
 		message = "登录态已失效，任务未执行。"
 	}
+	skipReason, cancellationMessage := authorizationFailureFollowupCancellation(disposition.Code, message)
+	runtimeError["canceled"] = true
+	runtimeError["cancellation_reason"] = skipReason
+	runtimeError["source_authorization_failure_message"] = message
+	runtimeError["message"] = cancellationMessage
+	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 cancellation payload")
+	}
 	executionOwner := strings.TrimSpace(task.ExecutionOwner)
 	if executionOwner == "" {
 		executionOwner = "legacy"
@@ -3282,28 +3287,28 @@ func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorization
 
 	tag, err := tx.Exec(ctx, `
 		UPDATE monitoring_collect_tasks
-		SET status = 'failed',
+		SET status = 'skipped',
 		    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,
+		    skip_reason = $8,
+		    error_message = $9,
 		    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,
+			) || $10::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 COALESCE(execution_owner, 'legacy') = $11
 		  AND id <> $5
 		  AND status IN ('pending', 'leased', 'expired')
 		  AND callback_received_at IS NULL
@@ -3311,9 +3316,9 @@ func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorization
 		    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)
+	`, task.TenantID, task.CollectorType, task.BusinessDate.Format("2006-01-02"), task.AIPlatformID, task.ID, targetClientID, completedAt, skipReason, cancellationMessage, requestPayloadJSON, executionOwner)
 	if err != nil {
-		return 0, response.ErrInternal(50041, "task_update_failed", "failed to fail remaining monitoring tasks after authorization failure")
+		return 0, response.ErrInternal(50041, "task_update_failed", "failed to skip remaining monitoring tasks after authorization failure")
 	}
 	return tag.RowsAffected(), nil
 }

server/internal/tenant/app/monitoring_callback_service_test.go

@@ -10,10 +10,12 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/geo-platform/tenant-api/internal/shared/config"
+	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
 
 func TestMonitoringAnswerParseModelUsesConfig(t *testing.T) {
@@ -635,6 +637,109 @@ func TestMonitoringAuthorizationFailureTargetClientPrefersExplicitTargetClient(t
 	}
 }
 
+func TestMonitorAuthorizationFollowupCancellationIsPlatformScopedAcrossAccounts(t *testing.T) {
+	tx := &captureDailyMonitorTaskTx{}
+	task := &repository.DesktopTask{
+		DesktopID:       uuid.New(),
+		TenantID:        77,
+		WorkspaceID:     88,
+		Kind:            "monitor",
+		TargetClientID:  uuid.New(),
+		TargetAccountID: uuid.New(),
+		Platform:        "doubao",
+	}
+
+	rows, err := loadQueuedDesktopAuthorizationFailureCandidates(context.Background(), tx, task)
+
+	require.NoError(t, err)
+	assert.Empty(t, rows)
+	require.True(t, tx.queryCalled)
+	query := normalizeSQLWhitespace(tx.querySQL)
+	assert.Contains(t, query, "AND kind = $3")
+	assert.Contains(t, query, "AND target_client_id = $4")
+	assert.Contains(t, query, "AND (kind = 'monitor' OR target_account_id = $5)")
+	assert.Contains(t, query, "AND platform_id = $6")
+	assert.Equal(t, "aborted", desktopAuthorizationFollowupTaskStatus("monitor"))
+	assert.Equal(t, "failed", desktopAuthorizationFollowupTaskStatus("publish"))
+	assert.Equal(t, "task_canceled", desktopAuthorizationFollowupEventType(&repository.DesktopTask{
+		Kind:   "monitor",
+		Status: "aborted",
+	}))
+}
+
+func TestFailRemainingMonitoringTasksAfterFinalAccountChallengeScopesToModelAndClient(t *testing.T) {
+	targetClientID := uuid.New().String()
+	completedAt := time.Date(2026, 7, 14, 9, 30, 0, 0, time.UTC)
+	message := "豆包账号触发人机验证，请完成验证码后再继续采集。"
+	tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("UPDATE 3")}
+	task := &monitoringCollectTask{
+		ID:             901,
+		TenantID:       77,
+		AIPlatformID:   "doubao",
+		CollectorType:  monitoringCollectorType,
+		BusinessDate:   time.Date(2026, 7, 14, 0, 0, 0, 0, time.UTC),
+		ExecutionOwner: "desktop_tasks",
+		TargetClientID: sql.NullString{String: targetClientID, Valid: true},
+	}
+	req := MonitoringTaskResultRequest{
+		Status:       "failed",
+		ErrorMessage: &message,
+		RawResponseJSON: map[string]interface{}{
+			"runtime_error": map[string]interface{}{
+				"code":             "doubao_challenge_required",
+				"message":          message,
+				"retryable":        false,
+				"non_retryable":    true,
+				"failure_category": "ai_platform_authorization",
+			},
+		},
+	}
+
+	affected, err := (&MonitoringCallbackService{}).failRemainingMonitoringTasksForAuthorizationFailure(
+		context.Background(),
+		tx,
+		task,
+		req,
+		completedAt,
+	)
+
+	require.NoError(t, err)
+	assert.Equal(t, int64(3), affected)
+	require.True(t, tx.execCalled)
+	query := normalizeSQLWhitespace(tx.execSQL)
+	assert.Contains(t, query, "SET status = 'skipped'")
+	assert.Contains(t, query, "skip_reason = $8")
+	assert.Contains(t, query, "WHERE tenant_id = $1")
+	assert.Contains(t, query, "AND business_date = $3::date")
+	assert.Contains(t, query, "AND ai_platform_id = $4")
+	assert.Contains(t, query, "AND id <> $5")
+	assert.Contains(t, query, "target_client_id::text = $6 OR leased_to_executor = $6")
+	assert.Contains(t, query, "AND status IN ('pending', 'leased', 'expired')")
+	require.Len(t, tx.execArgs, 11)
+	assert.Equal(t, int64(77), tx.execArgs[0])
+	assert.Equal(t, monitoringCollectorType, tx.execArgs[1])
+	assert.Equal(t, "2026-07-14", tx.execArgs[2])
+	assert.Equal(t, "doubao", tx.execArgs[3])
+	assert.Equal(t, int64(901), tx.execArgs[4])
+	assert.Equal(t, targetClientID, tx.execArgs[5])
+	assert.Equal(t, completedAt, tx.execArgs[6])
+	assert.Equal(t, monitoringPlatformHumanVerificationSkipReason, tx.execArgs[7])
+	assert.Equal(t, "该模型没有其他可用账号，因触发人机验证，后续采集任务已自动取消。", tx.execArgs[8])
+	assert.Equal(t, "desktop_tasks", tx.execArgs[10])
+
+	payloadJSON, ok := tx.execArgs[9].([]byte)
+	require.True(t, ok)
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(payloadJSON, &payload))
+	assert.Equal(t, true, payload["blocked_by_authorization_failure"])
+	runtimeError, ok := payload["runtime_error"].(map[string]interface{})
+	require.True(t, ok)
+	assert.Equal(t, "doubao_challenge_required", runtimeError["code"])
+	assert.Equal(t, "doubao", runtimeError["blocked_by_authorization_failure_platform"])
+	assert.Equal(t, true, runtimeError["canceled"])
+	assert.Equal(t, monitoringPlatformHumanVerificationSkipReason, runtimeError["cancellation_reason"])
+}
+
 func TestBuildMonitoringRawPayloadQwenIncludesSearchResultsInCitationInputs(t *testing.T) {
 	searchTitle := "搜索网页结果"