diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 7888173..6e15392 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -29,8 +29,8 @@ import { forgetTrackedAccountHealth, getAccountHealthSnapshot, getProjectedAccountHealth, - markTrackedAccountMissingPartition, markTrackedAccountBound, + markTrackedAccountMissingPartition, probeTrackedAccount, reportAccountFailure, syncTrackedAccounts, @@ -407,6 +407,137 @@ function accountActionRequiredSummary( return `${label} 账号需要完成人机验证后才能继续执行。` } +function markAuthorizationFailureNonRetryable( + error: Record, + category: 'ai_platform_authorization' | 'media_account_authorization', +): Record { + return { + ...error, + retryable: false, + non_retryable: true, + failure_category: category, + } +} + +function isAuthorizationFailureCode(code: string | null | undefined): boolean { + if (!code) { + return false + } + const normalized = code.trim().toLowerCase() + return ( + normalized === 'desktop_account_auth_expired' || + normalized === 'desktop_account_challenge_required' || + normalized === 'desktop_account_risk_control' || + normalized.endsWith('_not_logged_in') || + normalized.endsWith('_login_required') || + normalized.endsWith('_login_expired') || + normalized.endsWith('_challenge_required') + ) +} + +function isAuthorizationFailureMessage(message: string | null | undefined): boolean { + if (!message) { + return false + } + const normalized = message.trim().toLowerCase() + return ( + normalized.includes('login required') || + normalized.includes('login expired') || + normalized.includes('not logged in') || + normalized.includes('unauthorized') || + normalized.includes('登录态失效') || + normalized.includes('登录态已失效') || + normalized.includes('登录已失效') || + normalized.includes('请重新登录') || + normalized.includes('请先登录') || + normalized.includes('未登录') + ) +} + +function isAuthorizationFailureCategory(category: string | null | undefined): boolean { + return category === 'ai_platform_authorization' || category === 'media_account_authorization' +} + +function isNonRetryableAuthorizationFailure( + task: RuntimeTaskRecord, + result: AdapterExecutionResult, +): boolean { + if (result.status !== 'failed') { + return false + } + + const code = typeof result.error?.code === 'string' ? result.error.code : null + const message = typeof result.error?.message === 'string' ? result.error.message : null + const category = + typeof result.error?.failure_category === 'string' ? result.error.failure_category : null + if (isAuthorizationFailureCategory(category)) { + return true + } + if (!isAuthorizationFailureCode(code) && !isAuthorizationFailureMessage(message)) { + return false + } + + return task.kind === 'publish' || isAIPlatformId(task.platform) +} + +function withAuthorizationRetryPolicy( + task: RuntimeTaskRecord, + result: AdapterExecutionResult, +): AdapterExecutionResult { + if (!isNonRetryableAuthorizationFailure(task, result)) { + return result + } + + const category = isAIPlatformId(task.platform) + ? 'ai_platform_authorization' + : 'media_account_authorization' + const baseError = result.error ?? { + code: 'desktop_account_authorization_failed', + message: result.summary, + } + return { + ...result, + error: markAuthorizationFailureNonRetryable(baseError, category), + } +} + +async function resolveAuthorizationPreflightBlock( + task: RuntimeTaskRecord, +): Promise { + const accountIdentity = accountIdentityFromTask(task) + if (!accountIdentity || !shouldEnforceActiveAccount(task)) { + return null + } + + const previousSnapshot = getAccountHealthSnapshot(accountIdentity.id) + const readiness = await ensureAccountReady(accountIdentity) + if (readiness.authState === 'active') { + return null + } + + const result = withAuthorizationRetryPolicy(task, buildAccountBlockedResult(task, readiness)) + maybeRecordAccountAccessAlert(task, accountIdentity, previousSnapshot, readiness) + enqueueAccountHealthReport(accountIdentity.id) + recordAuthorizationPreflightBlock(task, result) + return result +} + +function recordAuthorizationPreflightBlock( + task: RuntimeTaskRecord, + result: AdapterExecutionResult, +): void { + if (!isNonRetryableAuthorizationFailure(task, result)) { + return + } + + const scope = task.kind === 'monitor' && isAIPlatformId(task.platform) ? '同模型' : '同账号' + recordActivity( + 'warn', + '授权失效，任务未执行', + `${runtimePlatformLabel(task.platform)} ${scope}授权不是 active，${task.title} 已直接回写 non-retryable 授权失败。`, + ) +} + function maybeRecordAccountAccessAlert( task: RuntimeTaskRecord, accountIdentity: PublishAccountIdentity | null, @@ -1363,7 +1494,8 @@ export function noteRuntimeAccountBound(account: DesktopAccountInfo): void { const existingIndex = state.accounts.findIndex( (item) => item.id === normalizedAccount.id || - (isAIPlatformId(normalizedAccount.platform) && item.platform === normalizedAccount.platform) || + (isAIPlatformId(normalizedAccount.platform) && + item.platform === normalizedAccount.platform) || (item.platform === normalizedAccount.platform && sameAccountPlatformUid(item.platform_uid, normalizedAccount.platform_uid)), ) @@ -1804,7 +1936,12 @@ async function executeLeasedMonitoringTask( queueMicrotask(() => pumpExecutionLoop()) try { - const execution = await executeTaskAdapter(taskRecord, abortController.signal) + const execution = + (await resolveAuthorizationPreflightBlock(taskRecord)) ?? + withAuthorizationRetryPolicy( + taskRecord, + await executeTaskAdapter(taskRecord, abortController.signal), + ) if (execution.status === 'unknown') { const skipReason = monitoringSkipReasonFromExecution(execution) await skipMonitoringTask(monitoringTaskID, { @@ -1937,6 +2074,14 @@ function buildMonitoringResultPayload( ): MonitoringTaskResultPayload { const payload = execution.payload ?? {} const status = execution.status === 'succeeded' ? 'succeeded' : 'failed' + const rawResponseJSON = asRecord(payload.raw_response_json) + const failedRawResponseJSON = + status === 'failed' && execution.error + ? { + ...rawResponseJSON, + runtime_error: execution.error, + } + : rawResponseJSON return { lease_token: leaseToken, status, @@ -1944,7 +2089,7 @@ function buildMonitoringResultPayload( provider_request_id: asOptionalString(payload.provider_request_id), request_id: asOptionalString(payload.request_id), answer: asOptionalString(payload.answer), - raw_response_json: asRecord(payload.raw_response_json), + raw_response_json: failedRawResponseJSON, citations: asSourceItems(payload.citations), search_results: asSourceItems(payload.search_results), brand_mentioned: asOptionalBoolean(payload.brand_mentioned), @@ -1961,12 +2106,20 @@ function buildMonitoringFailurePayload( leaseToken: string, failureMessage: string, structuredError: Record, + task?: RuntimeTaskRecord, ): MonitoringTaskResultPayload { + const runtimeError = task + ? (withAuthorizationRetryPolicy(task, { + status: 'failed', + summary: failureMessage, + error: structuredError, + }).error ?? structuredError) + : structuredError return { lease_token: leaseToken, status: 'failed', raw_response_json: { - runtime_error: structuredError, + runtime_error: runtimeError, }, error_message: failureMessage, completed_at: new Date().toISOString(), @@ -2007,7 +2160,9 @@ async function executeLeasedDesktopMonitorTask( } try { - const execution = await executeTaskAdapter(taskRecord, signal) + const execution = + (await resolveAuthorizationPreflightBlock(taskRecord)) ?? + withAuthorizationRetryPolicy(taskRecord, await executeTaskAdapter(taskRecord, signal)) if (execution.status === 'unknown') { const skipReason = monitoringSkipReasonFromExecution(execution) await skipMonitoringTask(monitoringTaskID, { @@ -2172,6 +2327,7 @@ async function executeLeasedDesktopMonitorTask( leased.lease_token as string, failureMessage || '监控任务执行失败。', structuredError, + taskRecord, ), ) } catch (submitError) { @@ -2402,7 +2558,12 @@ async function executeLeasedTask( return } - const execution = await executeTaskAdapter(taskRecord, abortController.signal) + const execution = + (await resolveAuthorizationPreflightBlock(taskRecord)) ?? + withAuthorizationRetryPolicy( + taskRecord, + await executeTaskAdapter(taskRecord, abortController.signal), + ) const completed = await completeDesktopTask(taskRecord.id, { lease_token: leased.lease_token, status: execution.status, @@ -2679,16 +2840,19 @@ function buildAccountBlockedResult( return { status: 'failed', summary: accountActionRequiredSummary(task.platform, readiness.authReason), - error: { - code: - readiness.authReason === 'risk_control' - ? 'desktop_account_risk_control' - : 'desktop_account_challenge_required', - message: - readiness.authReason === 'risk_control' - ? 'desktop account risk control triggered' - : 'desktop account challenge required', - }, + error: markAuthorizationFailureNonRetryable( + { + code: + readiness.authReason === 'risk_control' + ? 'desktop_account_risk_control' + : 'desktop_account_challenge_required', + message: + readiness.authReason === 'risk_control' + ? 'desktop account risk control triggered' + : 'desktop account challenge required', + }, + isAIPlatformId(task.platform) ? 'ai_platform_authorization' : 'media_account_authorization', + ), } } @@ -2696,10 +2860,13 @@ function buildAccountBlockedResult( return { status: 'failed', summary: `${task.accountName} 登录态已失效，请重新授权后再试。`, - error: { - code: 'desktop_account_auth_expired', - message: 'desktop account auth expired', - }, + error: markAuthorizationFailureNonRetryable( + { + code: 'desktop_account_auth_expired', + message: 'desktop account auth expired', + }, + isAIPlatformId(task.platform) ? 'ai_platform_authorization' : 'media_account_authorization', + ), } } @@ -2737,6 +2904,9 @@ async function maybeReportTaskAuthFailure( }) maybeRecordAccountAccessAlert(task, accountIdentity, previousSnapshot, nextSnapshot) + if (nextSnapshot && nextSnapshot.authState !== 'active') { + enqueueAccountHealthReport(accountIdentity.id) + } } function resolvePlaywrightTargetURL(platform: string, payload: Record): string { diff --git a/server/internal/tenant/app/desktop_task_authorization_failure.go b/server/internal/tenant/app/desktop_task_authorization_failure.go new file mode 100644 index 0000000..e22f44d --- /dev/null +++ b/server/internal/tenant/app/desktop_task_authorization_failure.go @@ -0,0 +1,420 @@ +package app + +import ( + "context" + "encoding/json" + "strings" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + "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 desktopAuthorizationFailureDisposition struct { + NonRetryable bool + Category string + Code string + Message string +} + +type desktopAuthorizationBulkFailure struct { + Tasks []*repository.DesktopTask + PublishOutcomes []*desktopPublishSyncOutcome + MonitorTaskIDs []int64 + ErrorPayload map[string]any +} + +type queuedDesktopAuthorizationFailureCandidate struct { + DesktopID uuid.UUID + MonitorTaskID *int64 +} + +func bulkFailQueuedDesktopTasksForAuthorizationFailure( + ctx context.Context, + tx pgx.Tx, + task *repository.DesktopTask, + errorJSON []byte, +) (*desktopAuthorizationBulkFailure, error) { + result := &desktopAuthorizationBulkFailure{} + if tx == nil || task == nil || task.Status != "failed" { + return result, nil + } + + errorPayload := unmarshalJSONObject(errorJSON) + disposition, ok := desktopAuthorizationFailureDispositionFromPayload(task, errorPayload) + if !ok || !disposition.NonRetryable { + return result, nil + } + result.ErrorPayload = ensureDesktopAuthorizationFailurePayload(errorPayload, disposition, task, nil) + + candidates, err := loadQueuedDesktopAuthorizationFailureCandidates(ctx, tx, task) + if err != nil { + return nil, err + } + for _, candidate := range candidates { + 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 + + 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) + if updateErr != nil { + return nil, updateErr + } + if updated == nil { + continue + } + result.Tasks = append(result.Tasks, updated) + + if updated.Kind == "publish" { + publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, updated) + if syncErr != nil { + return nil, syncErr + } + if publishOutcome != nil { + result.PublishOutcomes = append(result.PublishOutcomes, publishOutcome) + } + } + if updated.Kind == "monitor" && updated.MonitorTaskID != nil { + result.MonitorTaskIDs = append(result.MonitorTaskIDs, *updated.MonitorTaskID) + } + } + + return result, nil +} + +func loadQueuedDesktopAuthorizationFailureCandidates( + ctx context.Context, + tx pgx.Tx, + task *repository.DesktopTask, +) ([]queuedDesktopAuthorizationFailureCandidate, error) { + rows, err := tx.Query(ctx, ` + SELECT desktop_id, monitor_task_id + FROM desktop_tasks + WHERE tenant_id = $1 + AND workspace_id = $2 + AND kind = $3 + AND status = 'queued' + AND target_client_id = $4 + AND 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 + FOR UPDATE SKIP LOCKED + `, task.TenantID, task.WorkspaceID, task.Kind, task.TargetClientID, task.TargetAccountID, task.Platform, task.DesktopID) + if err != nil { + return nil, response.ErrInternal(50121, "desktop_task_authorization_failure_query_failed", "failed to select queued desktop tasks blocked by authorization failure") + } + defer rows.Close() + + candidates := make([]queuedDesktopAuthorizationFailureCandidate, 0) + for rows.Next() { + var ( + candidate queuedDesktopAuthorizationFailureCandidate + monitorTaskID pgtype.Int8 + ) + if scanErr := rows.Scan(&candidate.DesktopID, &monitorTaskID); scanErr != nil { + return nil, response.ErrInternal(50121, "desktop_task_authorization_failure_scan_failed", "failed to parse queued desktop tasks blocked by authorization failure") + } + if monitorTaskID.Valid { + value := monitorTaskID.Int64 + candidate.MonitorTaskID = &value + } + candidates = append(candidates, candidate) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50121, "desktop_task_authorization_failure_scan_failed", "failed to iterate queued desktop tasks blocked by authorization failure") + } + return candidates, nil +} + +func failQueuedDesktopTaskForAuthorizationFailure( + ctx context.Context, + tx pgx.Tx, + desktopID uuid.UUID, + errorJSON []byte, +) (*repository.DesktopTask, error) { + row := tx.QueryRow(ctx, ` + UPDATE desktop_tasks AS t + SET status = 'failed', + result = NULL, + error = $2, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE desktop_id = $1 + AND status = 'queued' + RETURNING `+desktopTaskRepositoryReturningColumns, + desktopID, + errorJSON, + ) + updated, err := scanRepositoryDesktopTask(row) + if err != nil { + 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 updated, nil +} + +func desktopAuthorizationFailureDispositionFromPayload( + task *repository.DesktopTask, + payload map[string]any, +) (desktopAuthorizationFailureDisposition, bool) { + if task == nil || len(payload) == 0 { + return desktopAuthorizationFailureDisposition{}, false + } + + disposition := desktopAuthorizationFailureDisposition{ + Category: desktopAuthorizationFailureString(payload["failure_category"]), + Code: desktopAuthorizationFailureString(payload["code"]), + Message: desktopAuthorizationFailureString(payload["message"]), + } + hasAuthorizationSignal := desktopAuthorizationFailureCategory(disposition.Category) || + desktopAuthorizationFailureCode(disposition.Code) || + desktopAuthorizationFailureMessage(disposition.Message) + if !hasAuthorizationSignal { + return desktopAuthorizationFailureDisposition{}, false + } + + disposition.NonRetryable = true + if disposition.Category == "" { + disposition.Category = desktopAuthorizationFailureCategoryForTask(task) + } + if !desktopAuthorizationFailureCategoryMatchesTask(task, disposition.Category) { + return desktopAuthorizationFailureDisposition{}, false + } + return disposition, true +} + +func ensureDesktopAuthorizationFailurePayload( + payload map[string]any, + disposition desktopAuthorizationFailureDisposition, + task *repository.DesktopTask, + monitorTaskID *int64, +) map[string]any { + result := make(map[string]any, len(payload)+6) + for key, value := range payload { + result[key] = value + } + if disposition.Code != "" { + result["code"] = disposition.Code + } else if _, exists := result["code"]; !exists { + result["code"] = "desktop_account_auth_expired" + } + if disposition.Message != "" { + result["message"] = disposition.Message + } else if _, exists := result["message"]; !exists { + result["message"] = "登录态已失效，任务未执行。" + } + result["retryable"] = false + result["non_retryable"] = true + if disposition.Category != "" { + result["failure_category"] = disposition.Category + } else if task != nil { + result["failure_category"] = desktopAuthorizationFailureCategoryForTask(task) + } + if task != nil && strings.TrimSpace(task.Platform) != "" { + result["platform"] = strings.TrimSpace(task.Platform) + } + if monitorTaskID != nil { + result["monitoring_task_id"] = *monitorTaskID + } + return result +} + +func desktopAuthorizationFailureCategoryForTask(task *repository.DesktopTask) string { + if task != nil && task.Kind == "monitor" { + return "ai_platform_authorization" + } + return "media_account_authorization" +} + +func desktopAuthorizationFailureCategoryMatchesTask(task *repository.DesktopTask, category string) bool { + category = strings.TrimSpace(category) + if task != nil && task.Kind == "monitor" { + return category == "ai_platform_authorization" + } + if task != nil && task.Kind == "publish" { + return category == "media_account_authorization" + } + return false +} + +func desktopAuthorizationFailureCategory(category string) bool { + category = strings.TrimSpace(category) + return category == "ai_platform_authorization" || category == "media_account_authorization" +} + +func desktopAuthorizationFailureCode(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 desktopAuthorizationFailureMessage(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 desktopAuthorizationFailureString(value any) string { + text, ok := value.(string) + if !ok { + return "" + } + return strings.TrimSpace(text) +} + +func failMonitoringCollectTasksForDesktopAuthorizationFailure( + ctx context.Context, + pool *pgxpool.Pool, + monitorTaskIDs []int64, + errorPayload map[string]any, +) error { + if pool == nil || len(monitorTaskIDs) == 0 { + return nil + } + + errorMessage := desktopAuthorizationFailureString(errorPayload["message"]) + if errorMessage == "" { + errorMessage = "登录态已失效，任务未执行。" + } + requestPayloadJSON, err := json.Marshal(map[string]any{ + "runtime_error": errorPayload, + "blocked_by_authorization_failure": true, + }) + if err != nil { + return response.ErrInternal(50124, "monitoring_authorization_failure_payload_failed", "failed to encode monitoring authorization failure payload") + } + + if _, err := pool.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 = NOW(), + skip_reason = NULL, + error_message = $2, + 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, + 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") + } + return nil +} + +func publishDesktopTaskLifecycleEvent( + ctx context.Context, + messaging *rabbitmq.Client, + logger *zap.Logger, + task *repository.DesktopTask, + eventType string, +) { + if task == nil { + return + } + + title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload( + task.Kind, + task.Payload, + ) + event := DesktopTaskEvent{ + Type: eventType, + TaskID: task.DesktopID.String(), + JobID: task.JobID.String(), + WorkspaceID: task.WorkspaceID, + TargetAccountID: task.TargetAccountID.String(), + TargetClientID: task.TargetClientID.String(), + 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 err := publishDesktopTaskEvent(ctx, messaging, event); err != nil && logger != nil { + logger.Warn("desktop task event publish failed", + zap.String("task_id", task.DesktopID.String()), + zap.String("event_type", eventType), + zap.Error(err), + ) + } + + if err := publishDesktopDispatchEvent(ctx, messaging, logger, 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 && logger != nil { + logger.Warn("desktop dispatch mirror publish failed", + zap.String("task_id", task.DesktopID.String()), + zap.String("event_type", eventType), + zap.Error(err), + ) + } +} diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 694f6f6..4fd2eda 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -20,7 +20,6 @@ import ( "github.com/geo-platform/tenant-api/internal/shared/config" "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" tenantcompliance "github.com/geo-platform/tenant-api/internal/tenant/compliance" "github.com/geo-platform/tenant-api/internal/tenant/repository" ) @@ -702,6 +701,10 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De if err != nil { return nil, err } + bulkFailure, err := bulkFailQueuedDesktopTasksForAuthorizationFailure(ctx, tx, task, errorJSON) + if err != nil { + return nil, err + } if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50104, "desktop_task_complete_commit_failed", "failed to commit desktop task completion") @@ -713,8 +716,19 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De s.logWarn("monitoring article alias sync failed after publish complete", syncErr, zap.Int64("article_id", publishOutcome.ArticleID)) } } + for _, outcome := range bulkFailure.PublishOutcomes { + invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID) + if outcome.ArticleAlias != nil { + if syncErr := s.syncMonitoringArticleAlias(ctx, outcome.ArticleAlias); syncErr != nil { + s.logWarn("monitoring article alias sync failed after bulk authorization failure", syncErr, zap.Int64("article_id", outcome.ArticleID)) + } + } + } s.publishTaskEvent(ctx, task, "task_completed") + for _, failedTask := range bulkFailure.Tasks { + s.publishTaskEvent(ctx, failedTask, "task_completed") + } view := buildDesktopTaskView(task) return &view, nil @@ -1760,56 +1774,7 @@ func requeuePreemptedMonitorTask( } func (s *DesktopTaskService) publishTaskEvent(ctx context.Context, task *repository.DesktopTask, eventType string) { - if task == nil { - return - } - - title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload) - event := DesktopTaskEvent{ - Type: eventType, - TaskID: task.DesktopID.String(), - JobID: task.JobID.String(), - WorkspaceID: task.WorkspaceID, - TargetAccountID: task.TargetAccountID.String(), - TargetClientID: task.TargetClientID.String(), - 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, - } - err := publishDesktopTaskEvent(ctx, s.messaging, event) - if err != nil { - s.logWarn("desktop task event publish failed", err, zap.String("task_id", task.DesktopID.String()), zap.String("event_type", eventType)) - } - - if dispatchErr := publishDesktopDispatchEvent(ctx, s.messaging, s.logger, 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, - }); dispatchErr != nil { - s.logWarn("desktop dispatch mirror publish failed", dispatchErr, zap.String("task_id", task.DesktopID.String()), zap.String("event_type", eventType)) - } + publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, task, eventType) } func (s *DesktopTaskService) logWarn(message string, err error, fields ...zap.Field) { diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index fd26640..0f43eae 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/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" } diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index 27b75fb..70ddb7a 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -1,11 +1,14 @@ package app import ( + "database/sql" "encoding/json" "errors" "strings" "testing" "time" + + "github.com/google/uuid" ) func TestDecideHeartbeatPrimaryLease(t *testing.T) { @@ -363,6 +366,89 @@ func TestValidateMonitoringTaskResultSubmissionRequiresSourcesForCitationMarkers } } +func TestMonitoringTaskReadyForQueuedResultAcceptsDesktopReceivedStatus(t *testing.T) { + if !monitoringTaskReadyForQueuedResult(&monitoringCollectTask{ + ExecutionOwner: "desktop_tasks", + Status: "received", + CallbackReceivedAt: sqlNullTimeForTest(), + }) { + t.Fatal("monitoringTaskReadyForQueuedResult() desktop received = false, want true") + } + + if !monitoringTaskReadyForQueuedResult(&monitoringCollectTask{ + ExecutionOwner: "desktop_tasks", + Status: "pending", + CallbackReceivedAt: sqlNullTimeForTest(), + }) { + t.Fatal("monitoringTaskReadyForQueuedResult() legacy desktop pending callback = false, want true") + } +} + +func sqlNullTimeForTest() sql.NullTime { + return sql.NullTime{Time: time.Now(), Valid: true} +} + +func TestBuildDesktopMonitorTaskErrorValuePreservesNonRetryableAuthorizationFailure(t *testing.T) { + message := "用户登录态已失效，请重新授权后再试。" + value := buildDesktopMonitorTaskErrorValue(&monitoringCollectTask{ + ID: 522, + AIPlatformID: "yuanbao", + }, MonitoringTaskResultRequest{ + Status: "failed", + ErrorMessage: &message, + RawResponseJSON: map[string]interface{}{ + "runtime_error": map[string]interface{}{ + "code": "desktop_account_auth_expired", + "message": message, + "retryable": false, + "failure_category": "ai_platform_authorization", + }, + }, + }) + + if value["code"] != "desktop_account_auth_expired" { + t.Fatalf("error code = %v, want desktop_account_auth_expired", value["code"]) + } + if value["retryable"] != false { + t.Fatalf("retryable = %v, want false", value["retryable"]) + } + if value["failure_category"] != "ai_platform_authorization" { + t.Fatalf("failure_category = %v, want ai_platform_authorization", value["failure_category"]) + } +} + +func TestMonitoringTaskFailureDispositionDetectsAuthorizationFailureFromMessage(t *testing.T) { + message := "登录态已失效，请重新授权后再试。" + got := monitoringTaskFailureDispositionFromRequest(MonitoringTaskResultRequest{ + Status: "failed", + ErrorMessage: &message, + RawResponseJSON: map[string]interface{}{ + "runtime_error": map[string]interface{}{ + "message": message, + }, + }, + }) + + if !got.NonRetryable { + t.Fatal("NonRetryable = false, want true") + } + if got.Category != "ai_platform_authorization" { + t.Fatalf("Category = %q, want ai_platform_authorization", got.Category) + } +} + +func TestMonitoringAuthorizationFailureTargetClientPrefersExplicitTargetClient(t *testing.T) { + targetClientID := uuid.New().String() + task := &monitoringCollectTask{ + TargetClientID: sql.NullString{String: targetClientID, Valid: true}, + LeasedToExecutor: sql.NullString{String: uuid.New().String(), Valid: true}, + } + + if got := monitoringAuthorizationFailureTargetClientID(task); got != targetClientID { + t.Fatalf("target client = %q, want %q", got, targetClientID) + } +} + func TestBuildMonitoringRawPayloadQwenIncludesSearchResultsInCitationInputs(t *testing.T) { searchTitle := "搜索网页结果" diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 07673b9..8352026 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -924,6 +924,7 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( AND t.collector_type = $3 AND t.business_date = $4::date AND t.status = 'pending' + AND t.callback_received_at IS NULL AND t.trigger_source = 'automatic' AND t.run_mode = 'desktop_standard' AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks' diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 82bf28f..88e5db5 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -144,6 +144,7 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( AND t.collector_type = $2 AND t.business_date = $3::date AND t.status = 'pending' + AND t.callback_received_at IS NULL AND COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks' AND t.question_id = ANY($4) AND t.ai_platform_id = ANY($5)