From 6aea051c19b61f78f28b60c0d2e8b919d70532e1 Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 22 Jul 2026 23:57:28 +0800 Subject: [PATCH] feat(monitoring): requeue automatic tasks after a manual human-verification run succeeds When a manual, desktop-owned collect task succeeds after clearing a human verification gate, requeue the automatic monitoring tasks that were previously skipped for the same tenant/business-date/platform/client with the human-verification skip reason. Restored tasks are reset to pending (clearing lease, callback, and completion fields), stamped with a recovery marker in their payload, and their brand projections are rebuilt after commit. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tenant/app/monitoring_callback_service.go | 30 +++++- .../app/monitoring_callback_service_test.go | 69 +++++++++++++ .../monitoring_human_verification_recovery.go | 99 +++++++++++++++++++ 3 files changed, 196 insertions(+), 2 deletions(-) create mode 100644 server/internal/tenant/app/monitoring_human_verification_recovery.go diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index b7468cf..18cbf6e 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -1332,7 +1332,7 @@ func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context, return nil, nil } - return s.processTaskResult(ctx, task, envelope.CallbackResult) + return s.processTaskResult(ctx, task, envelope.CallbackResult, envelope.InstallationID) } func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID string, installationToken string) (*monitoringInstallationIdentity, error) { @@ -2327,7 +2327,12 @@ func (s *MonitoringCallbackService) upsertParseResult(ctx context.Context, tx pg return nil } -func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) { +func (s *MonitoringCallbackService) processTaskResult( + ctx context.Context, + task *monitoringCollectTask, + req MonitoringTaskResultRequest, + installationID string, +) (*MonitoringTaskResultResponse, error) { if task == nil || strings.TrimSpace(task.QuestionText) == "" { return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "monitoring task question text is empty") } @@ -2421,6 +2426,14 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task } } + recoveredBrandIDs := []int64(nil) + if shouldRecoverHumanVerificationTasks(task, req) { + recoveredBrandIDs, err = s.restoreHumanVerificationTasks(ctx, tx, task, installationID) + if err != nil { + return nil, err + } + } + if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil { return nil, err } @@ -2429,6 +2442,19 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest") } + recoveredProjectionBrandIDs := make(map[int64]struct{}, len(recoveredBrandIDs)) + for _, brandID := range recoveredBrandIDs { + if brandID > 0 && brandID != task.BrandID { + recoveredProjectionBrandIDs[brandID] = struct{}{} + } + } + s.dispatchMonitoringProjectionRebuilds( + task.TenantID, + task.BusinessDate, + recoveredProjectionBrandIDs, + "human_verification_recovery", + ) + var runIDPtr *int64 if !failedOutcome.Requeue { runIDPtr = &runID diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index 607d944..83624b3 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -763,6 +763,75 @@ func TestFailRemainingMonitoringTasksAfterFinalAccountChallengeScopesToModelAndC assert.Equal(t, monitoringPlatformHumanVerificationSkipReason, runtimeError["cancellation_reason"]) } +func TestShouldRecoverHumanVerificationTasksAfterManualSuccess(t *testing.T) { + task := &monitoringCollectTask{ + CollectorType: monitoringCollectorType, + ExecutionOwner: "desktop_tasks", + TriggerSource: "manual", + } + + assert.True(t, shouldRecoverHumanVerificationTasks(task, MonitoringTaskResultRequest{Status: "succeeded"})) + assert.False(t, shouldRecoverHumanVerificationTasks(task, MonitoringTaskResultRequest{Status: "failed"})) + + automatic := *task + automatic.TriggerSource = "automatic" + assert.False(t, shouldRecoverHumanVerificationTasks(&automatic, MonitoringTaskResultRequest{Status: "succeeded"})) +} + +func TestRestoreHumanVerificationTasksRequeuesOnlySkippedAutomaticTasks(t *testing.T) { + tx := &captureDailyMonitorTaskTx{} + task := &monitoringCollectTask{ + ID: 901, + TenantID: 77, + AIPlatformID: "doubao", + CollectorType: monitoringCollectorType, + BusinessDate: time.Date(2026, 7, 22, 0, 0, 0, 0, time.UTC), + ExecutionOwner: "desktop_tasks", + TriggerSource: "manual", + } + + brandIDs, err := (&MonitoringCallbackService{}).restoreHumanVerificationTasks( + context.Background(), + tx, + task, + "00000000-0000-0000-0000-000000000123", + ) + + require.NoError(t, err) + assert.Empty(t, brandIDs) + require.True(t, tx.queryCalled) + query := normalizeSQLWhitespace(tx.querySQL) + for _, fragment := range []string{ + "SET status = 'pending'", + "callback_received_at = NULL", + "completed_at = NULL", + "skip_reason = NULL", + "dispatch_after = NOW()", + "last_dispatched_at = NULL", + "tenant_id = $1", + "business_date = $2::date", + "ai_platform_id = $3", + "collector_type = $4", + "trigger_source = 'automatic'", + "status = 'skipped'", + "skip_reason = $6", + "target_client_id::text = $7", + "'recovered_by_monitoring_task_id', $8::bigint", + "RETURNING brand_id", + } { + assert.Contains(t, query, fragment) + } + require.Len(t, tx.queryArgs, 8) + assert.Equal(t, int64(77), tx.queryArgs[0]) + assert.Equal(t, "2026-07-22", tx.queryArgs[1]) + assert.Equal(t, "doubao", tx.queryArgs[2]) + assert.Equal(t, monitoringCollectorType, tx.queryArgs[3]) + assert.Equal(t, "desktop_tasks", tx.queryArgs[4]) + assert.Equal(t, monitoringPlatformHumanVerificationSkipReason, tx.queryArgs[5]) + assert.Equal(t, "00000000-0000-0000-0000-000000000123", tx.queryArgs[6]) + assert.Equal(t, int64(901), tx.queryArgs[7]) +} + func TestBuildMonitoringRawPayloadQwenIncludesSearchResultsInCitationInputs(t *testing.T) { searchTitle := "搜索网页结果" diff --git a/server/internal/tenant/app/monitoring_human_verification_recovery.go b/server/internal/tenant/app/monitoring_human_verification_recovery.go new file mode 100644 index 0000000..010257a --- /dev/null +++ b/server/internal/tenant/app/monitoring_human_verification_recovery.go @@ -0,0 +1,99 @@ +package app + +import ( + "context" + "strings" + + "github.com/jackc/pgx/v5" + + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +func shouldRecoverHumanVerificationTasks( + task *monitoringCollectTask, + req MonitoringTaskResultRequest, +) bool { + return task != nil && + task.CollectorType == monitoringCollectorType && + strings.TrimSpace(task.ExecutionOwner) == "desktop_tasks" && + strings.TrimSpace(task.TriggerSource) == "manual" && + normalizeRunStatus(req.Status) == "succeeded" +} + +func (s *MonitoringCallbackService) restoreHumanVerificationTasks( + ctx context.Context, + tx pgx.Tx, + task *monitoringCollectTask, + installationID string, +) ([]int64, error) { + installationID = strings.TrimSpace(installationID) + if tx == nil || task == nil || installationID == "" { + return nil, nil + } + + rows, err := tx.Query(ctx, ` + UPDATE monitoring_collect_tasks + SET status = 'pending', + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + callback_received_at = NULL, + completed_at = NULL, + skip_reason = NULL, + error_message = NULL, + dispatch_after = NOW(), + last_dispatched_at = 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 + ) || jsonb_build_object( + 'human_verification_recovery', jsonb_build_object( + 'recovered_at', NOW(), + 'recovered_by_monitoring_task_id', $8::bigint + ) + ), + updated_at = NOW() + WHERE tenant_id = $1 + AND business_date = $2::date + AND ai_platform_id = $3 + AND collector_type = $4 + AND COALESCE(execution_owner, 'legacy') = $5 + AND trigger_source = 'automatic' + AND status = 'skipped' + AND skip_reason = $6 + AND target_client_id::text = $7 + RETURNING brand_id + `, + task.TenantID, + task.BusinessDate.Format("2006-01-02"), + task.AIPlatformID, + task.CollectorType, + "desktop_tasks", + monitoringPlatformHumanVerificationSkipReason, + installationID, + task.ID, + ) + if err != nil { + return nil, response.ErrInternal(50041, "human_verification_recovery_failed", "failed to restore monitoring tasks canceled by human verification") + } + defer rows.Close() + + brandIDs := make([]int64, 0) + for rows.Next() { + var brandID int64 + if scanErr := rows.Scan(&brandID); scanErr != nil { + return nil, response.ErrInternal(50041, "human_verification_recovery_failed", "failed to parse restored monitoring tasks") + } + if brandID > 0 { + brandIDs = append(brandIDs, brandID) + } + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "human_verification_recovery_failed", "failed to iterate restored monitoring tasks") + } + return uniqueSortedMonitoringProjectionBrandIDs(brandIDs), nil +}