fix: queue monitor tasks and recover doubao dom answers

server/internal/tenant/app/desktop_task_service.go

@@ -25,7 +25,10 @@ import (
 	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
 
-const desktopPublishMaxAttempts = 3
+const (
+	desktopPublishMaxAttempts                     = 3
+	maxConcurrentMonitorPlatformsPerDesktopClient = 2
+)
 
 var errDesktopTaskLeaseDeferred = errors.New("desktop task lease deferred")
 
@@ -520,7 +523,13 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 	}
 	hasAccountIDs := len(accountIDs) > 0
 
-	row := s.pool.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(),
+	tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	row := tx.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(),
 		client.TenantID,
 		client.WorkspaceID,
 		client.ID,
@@ -528,8 +537,16 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 		accountIDs,
 		params.AttemptID,
 		params.LeaseTokenHash,
+		maxConcurrentMonitorPlatformsPerDesktopClient,
 	)
-	return scanRepositoryDesktopTask(row)
+	task, err := scanRepositoryDesktopTask(row)
+	if err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop monitor task lease")
+	}
+	return task, nil
 }
 
 func leaseNextQueuedMonitorTaskSQL() string {
@@ -564,6 +581,16 @@ func leaseNextQueuedMonitorTaskSQL() string {
 				  AND recent.updated_at >= now() - interval '30 seconds'
 				  AND recent.desktop_id <> dt.desktop_id
 			  )
+			  AND (
+				SELECT COUNT(DISTINCT active_platform.platform_id)
+				FROM desktop_tasks AS active_platform
+				WHERE active_platform.tenant_id = dt.tenant_id
+				  AND active_platform.workspace_id = dt.workspace_id
+				  AND active_platform.kind = 'monitor'
+				  AND active_platform.target_client_id = $3
+				  AND active_platform.status = 'in_progress'
+				  AND active_platform.desktop_id <> dt.desktop_id
+			  ) < $8::integer
 			  AND (
 				dt.target_client_id = $3
 				OR ($4::boolean AND dt.target_account_id = ANY($5::uuid[]))
@@ -658,7 +685,13 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 	}
 	hasAccountIDs := len(accountIDs) > 0
 
-	row := s.pool.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(),
+	tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	row := tx.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(),
 		desktopID,
 		client.WorkspaceID,
 		client.TenantID,
@@ -668,8 +701,16 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 		params.AttemptID,
 		params.LeaseTokenHash,
 		desktopPublishMaxAttempts,
+		maxConcurrentMonitorPlatformsPerDesktopClient,
 	)
-	return scanRepositoryDesktopTask(row)
+	task, err := scanRepositoryDesktopTask(row)
+	if err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop task lease")
+	}
+	return task, nil
 }
 
 func leaseQueuedDesktopTaskByIDSQL() string {
@@ -722,6 +763,19 @@ func leaseQueuedDesktopTaskByIDSQL() string {
 					  AND recent.desktop_id <> dt.desktop_id
 				)
 			  )
+			  AND (
+				dt.kind <> 'monitor'
+				OR (
+					SELECT COUNT(DISTINCT active_platform.platform_id)
+					FROM desktop_tasks AS active_platform
+					WHERE active_platform.tenant_id = dt.tenant_id
+					  AND active_platform.workspace_id = dt.workspace_id
+					  AND active_platform.kind = 'monitor'
+					  AND active_platform.target_client_id = $4
+					  AND active_platform.status = 'in_progress'
+					  AND active_platform.desktop_id <> dt.desktop_id
+				) < $10::integer
+			  )
 			  AND (
 				(
 					dt.kind = 'monitor'
@@ -753,6 +807,21 @@ func leaseQueuedDesktopTaskByIDSQL() string {
 		RETURNING ` + desktopTaskRepositoryReturningColumns
 }
 
+func beginDesktopMonitorLeaseTx(ctx context.Context, pool *pgxpool.Pool, clientID uuid.UUID) (pgx.Tx, error) {
+	if pool == nil {
+		return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "desktop task pool is not available")
+	}
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to start desktop monitor task lease")
+	}
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(clientID)); err != nil {
+		_ = tx.Rollback(ctx)
+		return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to lock desktop monitor task lease")
+	}
+	return tx, nil
+}
+
 func (s *DesktopTaskService) loadMonitorLeaseAccountIDs(ctx context.Context, client *repository.DesktopClient) ([]uuid.UUID, error) {
 	if s == nil || client == nil || s.pool == nil {
 		return nil, nil

server/internal/tenant/app/desktop_task_service_test.go

@@ -249,6 +249,9 @@ func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T)
 		"active.platform_id = dt.platform_id",
 		"active.status = 'in_progress'",
 		"active.desktop_id <> dt.desktop_id",
+		"COUNT(DISTINCT active_platform.platform_id)",
+		"active_platform.target_client_id = $3",
+		"< $8::integer",
 		"recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')",
 		"recent.updated_at >= now() - interval '30 seconds'",
 	} {
@@ -268,6 +271,9 @@ func TestLeaseQueuedDesktopTaskByIDSQLSerializesMonitorPerClientPlatform(t *test
 		"active.platform_id = dt.platform_id",
 		"active.status = 'in_progress'",
 		"active.desktop_id <> dt.desktop_id",
+		"COUNT(DISTINCT active_platform.platform_id)",
+		"active_platform.target_client_id = $4",
+		"< $10::integer",
 		"recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')",
 		"recent.updated_at >= now() - interval '30 seconds'",
 	} {

server/internal/tenant/app/monitoring_phase2_desktop_tasks.go

@@ -246,14 +246,6 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks(
 			spec.TargetClientID = target.ClientID
 		}
 
-		available, throttleErr := acquireMonitorDesktopTaskPlatformSlot(ctx, tx, spec)
-		if throttleErr != nil {
-			return nil, nil, throttleErr
-		}
-		if !available {
-			continue
-		}
-
 		task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec)
 		if upsertErr != nil {
 			return nil, nil, upsertErr
@@ -282,43 +274,8 @@ func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) boo
 	return false
 }
 
-func acquireMonitorDesktopTaskPlatformSlot(ctx context.Context, tx pgx.Tx, spec monitorDesktopTaskSpec) (bool, error) {
-	if tx == nil || spec.TargetClientID == uuid.Nil || strings.TrimSpace(spec.PlatformID) == "" {
-		return false, nil
-	}
-	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskPlatformSlotLockKey(spec.TargetClientID, spec.PlatformID)); err != nil {
-		return false, response.ErrInternal(50123, "desktop_task_throttle_lock_failed", "failed to acquire phase2 monitor desktop platform slot")
-	}
-
-	var busy bool
-	if err := tx.QueryRow(ctx, monitorDesktopTaskPlatformSlotBusySQL(), spec.TenantID, spec.WorkspaceID, spec.TargetClientID, normalizeMonitoringPlatformID(spec.PlatformID), spec.MonitorTaskID).Scan(&busy); err != nil {
-		return false, response.ErrInternal(50124, "desktop_task_throttle_lookup_failed", "failed to inspect phase2 monitor desktop platform slot")
-	}
-	return !busy, nil
-}
-
-func monitorDesktopTaskPlatformSlotBusySQL() string {
-	return `
-		SELECT EXISTS (
-			SELECT 1
-			FROM desktop_tasks
-			WHERE tenant_id = $1
-			  AND workspace_id = $2
-			  AND kind = 'monitor'
-			  AND target_client_id = $3
-			  AND platform_id = $4
-			  AND status = 'in_progress'
-			  AND (
-				$5::bigint <= 0
-				OR monitor_task_id IS NULL
-				OR monitor_task_id <> $5
-			  )
-		)
-	`
-}
-
-func monitorDesktopTaskPlatformSlotLockKey(clientID uuid.UUID, platformID string) int64 {
-	return int64(monitoringDailyStableHash("monitor_desktop_platform_slot", clientID.String(), normalizeMonitoringPlatformID(platformID)))
+func monitorDesktopTaskClientSlotLockKey(clientID uuid.UUID) int64 {
+	return int64(monitoringDailyStableHash("monitor_desktop_client_slot", clientID.String()))
 }
 
 func (s *MonitoringService) loadMonitorDesktopTaskTargets(

server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go

@@ -207,21 +207,6 @@ func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) {
 	assert.Equal(t, "stale_business_date", outcome.StoredSkipReason)
 }
 
-func TestMonitorDesktopTaskPlatformSlotBusySQLThrottlesPerClientPlatformInProgress(t *testing.T) {
-	query := strings.Join(strings.Fields(monitorDesktopTaskPlatformSlotBusySQL()), " ")
-
-	for _, fragment := range []string{
-		"FROM desktop_tasks",
-		"kind = 'monitor'",
-		"target_client_id = $3",
-		"platform_id = $4",
-		"status = 'in_progress'",
-		"monitor_task_id <> $5",
-	} {
-		assert.Contains(t, query, fragment)
-	}
-}
-
 func TestRequestPhase2MonitorInterruptsAcceptsExcludedPlatforms(t *testing.T) {
 	clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701")
 	targets, err := (&MonitoringService{}).requestPhase2MonitorInterrupts(