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) <noreply@anthropic.com>
2026-07-22 23:57:28 +08:00
parent fc7495aba0
commit 6aea051c19
3 changed files with 196 additions and 2 deletions
@@ -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
@@ -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 := "搜索网页结果"

@@ -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
+}