diff --git a/server/internal/tenant/app/desktop_task_authorization_failure.go b/server/internal/tenant/app/desktop_task_authorization_failure.go index e22f44d..b922023 100644 --- a/server/internal/tenant/app/desktop_task_authorization_failure.go +++ b/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 } diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index aa99acb..bd8f103 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/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) diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 2684cf1..b7468cf 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/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 } diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index c6c8e1f..20eea5c 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/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 := "搜索网页结果"