fix: queue monitor tasks and recover doubao dom answers

apps/desktop-client/src/main/adapters/doubao.test.ts

@@ -251,10 +251,42 @@ describe('doubao adapter helpers', () => {
     })
   })
 
-  it('does not submit a success answer when the SSE stream has no answer text', () => {
+  it('uses DOM answer when Doubao finishes the stream without SSE answer text', () => {
     expect(
       resolveDoubaoSubmissionAnswer({
         answer: null,
+        domAnswer:
+          '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。',
+        allowDomAnswer: true,
+      }),
+    ).toEqual({
+      answer:
+        '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。',
+      source: 'dom',
+    })
+  })
+
+  it('does not submit DOM answer before a finish signal allows DOM recovery', () => {
+    expect(
+      resolveDoubaoSubmissionAnswer({
+        answer: null,
+        domAnswer:
+          '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。',
+        allowDomAnswer: false,
+      }),
+    ).toEqual({
+      answer: null,
+      source: null,
+    })
+  })
+
+  it('does not submit a weak history-title DOM fragment as a success answer', () => {
+    expect(
+      resolveDoubaoSubmissionAnswer({
+        answer: null,
+        domAnswer:
+          '口袋门墙体预埋件全解析 卧室口袋门阻尼缓冲导轨选购安装指南 口袋门防夹手缓冲方案 隐形门合页选购指南',
+        allowDomAnswer: true,
       }),
     ).toEqual({
       answer: null,

apps/desktop-client/src/main/adapters/doubao.ts

@@ -469,14 +469,45 @@ function selectBestDoubaoText(primary: string | null, secondary: string | null):
   return secondaryScore > primaryScore ? secondaryText : primaryText
 }
 
-function resolveDoubaoSubmissionAnswer(input: { answer: string | null }): {
+function isDoubaoSSESubmissionCandidate(value: string | null): boolean {
+  const answer = normalizeOptionalString(value)
+  if (!answer || containsDoubaoMojibakeSignal(answer)) {
+    return false
+  }
+  return true
+}
+
+function isDoubaoDOMSubmissionCandidate(value: string | null): boolean {
+  const answer = normalizeOptionalString(value)
+  if (!answer || containsDoubaoMojibakeSignal(answer)) {
+    return false
+  }
+  if (answer.length < 48) {
+    return false
+  }
+  const punctuationCount = countDoubaoMatches(answer, /[。！？；：，、,.!?;:]/g)
+  if (punctuationCount < 2 && !doubaoHasStructuredAnswerMarkers(answer) && !/\n/.test(answer)) {
+    return false
+  }
+  return doubaoTextQualityScore(answer) >= 180
+}
+
+function resolveDoubaoSubmissionAnswer(input: {
   answer: string | null
-  source: 'sse' | null
+  domAnswer?: string | null
+  allowDomAnswer?: boolean
+}): {
+  answer: string | null
+  source: 'sse' | 'dom' | null
 } {
   const answer = normalizeOptionalString(input.answer)
-  if (answer && !containsDoubaoMojibakeSignal(answer)) {
+  if (isDoubaoSSESubmissionCandidate(answer)) {
     return { answer, source: 'sse' }
   }
+  const domAnswer = normalizeOptionalString(input.domAnswer)
+  if (input.allowDomAnswer === true && isDoubaoDOMSubmissionCandidate(domAnswer)) {
+    return { answer: domAnswer, source: 'dom' }
+  }
 
   return { answer: null, source: null }
 }
@@ -2505,7 +2536,8 @@ const doubaoQueryInPage = async (
         return
       }
       seen.add(href)
-      const visibleText = normalize(element.innerText) ?? normalize(element.textContent) ?? fallbackText
+      const visibleText =
+        normalize(element.innerText) ?? normalize(element.textContent) ?? fallbackText
       links.push({
         url: href,
         title:
@@ -2643,9 +2675,7 @@ const doubaoQueryInPage = async (
     /搜索\s*\d+\s*个关键词|参考\s*\d+\s*篇资料|查看\d+篇资料|参考资料|资料来源/.test(line)
 
   const isReferenceQueryLine = (line: string): boolean =>
-    /[“"「].+[”"」]/.test(line) ||
-    /、/.test(line) ||
-    /关键词|搜索词|查询词/.test(line)
+    /[“"「].+[”"」]/.test(line) || /、/.test(line) || /关键词|搜索词|查询词/.test(line)
 
   const isReferenceItemLine = (line: string): boolean =>
     /^\d+[.、]\s*\S+/.test(line) &&
@@ -3240,6 +3270,8 @@ export const doubaoAdapter: MonitorAdapter = {
       pageResult.ok === false && isDoubaoRecoverablePageError(pageResult.error)
     const submission = resolveDoubaoSubmissionAnswer({
       answer,
+      domAnswer: pageResult.domAnswer,
+      allowDomAnswer: streamSawAnswerFinish || pageResult.ok === true,
     })
     const hasRecoveredAnswer = Boolean(submission.answer)
     const hasRecoveredContent = hasRecoveredAnswer || searchResults.length > 0

server/internal/tenant/app/desktop_task_service.go

@@ -25,7 +25,12 @@ 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")
 
 type DesktopTaskService struct {
 	pool           *pgxpool.Pool
@@ -247,7 +252,11 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
 			if taskID == nil {
 				return &LeaseDesktopTaskResponse{}, nil
 			}
-			return nil, s.classifyLeaseError(ctx, client, taskID)
+			classifiedErr := s.classifyLeaseError(ctx, client, taskID)
+			if errors.Is(classifiedErr, errDesktopTaskLeaseDeferred) {
+				return &LeaseDesktopTaskResponse{}, nil
+			}
+			return nil, classifiedErr
 		}
 		fields := []zap.Field{
 			zap.String("client_id", client.ID.String()),
@@ -514,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,
@@ -522,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 {
@@ -558,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[]))
@@ -652,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,
@@ -662,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 {
@@ -716,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'
@@ -747,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
@@ -966,6 +1041,9 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De
 	taskRepo := repository.NewDesktopTaskRepository(tx)
 	previousTask, _ := taskRepo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
 	previousAttemptID := activeAttemptID(previousTask)
+	if previousTask != nil {
+		finalStatus = normalizeDesktopTaskCompletionStatusForKind(previousTask.Kind, finalStatus)
+	}
 
 	task, err := taskRepo.Complete(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), finalStatus, resultJSON, errorJSON)
 	if err != nil {
@@ -1558,10 +1636,19 @@ func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *rep
 		return response.ErrForbidden(40381, "desktop_task_not_target_client", "desktop task is assigned to another desktop client")
 	}
 
+	return classifyDesktopTaskLeaseState(task)
+}
+
+func classifyDesktopTaskLeaseState(task *repository.DesktopTask) error {
+	if task == nil {
+		return response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
+	}
+	if task.Kind == "monitor" && task.Status == "queued" {
+		return errDesktopTaskLeaseDeferred
+	}
 	if task.Status == "in_progress" {
 		return response.ErrConflict(40989, "desktop_task_already_in_progress", "desktop task is already in progress")
 	}
-
 	return response.ErrConflict(40991, "desktop_task_not_queued", "desktop task is not queued")
 }
 
@@ -1646,13 +1733,22 @@ func normalizeDesktopTaskCompletionStatus(status string) string {
 	}
 }
 
-func normalizeDesktopTaskViewStatus(kind string, status string) string {
-	if strings.TrimSpace(kind) == "publish" {
-		return normalizeDesktopTaskTerminalStatus(status)
+func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) string {
+	if strings.TrimSpace(kind) == "monitor" && strings.TrimSpace(status) == "unknown" {
+		return "failed"
 	}
 	return strings.TrimSpace(status)
 }
 
+func normalizeDesktopTaskViewStatus(kind string, status string) string {
+	switch strings.TrimSpace(kind) {
+	case "monitor", "publish":
+		return normalizeDesktopTaskTerminalStatus(status)
+	default:
+		return strings.TrimSpace(status)
+	}
+}
+
 type desktopTaskRecoveryMode string
 
 const (

server/internal/tenant/app/desktop_task_service_test.go

@@ -1,11 +1,13 @@
 package app
 
 import (
+	"errors"
 	"regexp"
 	"strings"
 	"testing"
 
 	"github.com/geo-platform/tenant-api/internal/shared/response"
+	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
 
 func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing.T) {
@@ -138,19 +140,34 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) {
 	}
 }
 
-func TestNormalizeDesktopTaskCompletionStatusPreservesUnknown(t *testing.T) {
+func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testing.T) {
 	t.Parallel()
 
-	if got := normalizeDesktopTaskCompletionStatus("unknown"); got != "unknown" {
-		t.Fatalf("normalizeDesktopTaskCompletionStatus(unknown) = %q, want unknown", got)
+	if got := normalizeDesktopTaskCompletionStatusForKind("monitor", "unknown"); got != "failed" {
+		t.Fatalf("monitor completion status = %q, want failed", got)
+	}
+	if got := normalizeDesktopTaskCompletionStatusForKind("publish", "unknown"); got != "unknown" {
+		t.Fatalf("publish completion status = %q, want unknown", got)
 	}
 }
 
-func TestNormalizeDesktopTaskViewStatusOnlyMapsPublishUnknownToFailed(t *testing.T) {
+func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
 	t.Parallel()
 
-	if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "unknown" {
-		t.Fatalf("monitor unknown view status = %q, want unknown", got)
+	task := &repository.DesktopTask{
+		Kind:   "monitor",
+		Status: "queued",
+	}
+	if got := classifyDesktopTaskLeaseState(task); !errors.Is(got, errDesktopTaskLeaseDeferred) {
+		t.Fatalf("classifyDesktopTaskLeaseState() = %v, want deferred", got)
+	}
+}
+
+func TestNormalizeDesktopTaskViewStatusMapsMonitorAndPublishUnknownToFailed(t *testing.T) {
+	t.Parallel()
+
+	if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" {
+		t.Fatalf("monitor unknown view status = %q, want failed", got)
 	}
 	if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "failed" {
 		t.Fatalf("publish unknown view status = %q, want failed", got)
@@ -232,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'",
 	} {
@@ -251,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_callback_service.go

@@ -945,6 +945,7 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 	}
 
 	skipReason := normalizeSkipReason(req.SkipReason)
+	skipOutcome := monitoringSkipOutcomeForReason(skipReason)
 
 	tx, err := s.monitoringPool.Begin(ctx)
 	if err != nil {
@@ -954,14 +955,14 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 
 	if _, err := tx.Exec(ctx, `
 		UPDATE monitoring_collect_tasks
-		SET status = 'skipped',
+		SET status = $2,
 		    callback_received_at = NOW(),
-		    completed_at = $2,
-		    skip_reason = $3,
-		    error_message = $4,
+		    completed_at = $3,
+		    skip_reason = $4,
+		    error_message = $5,
 		    updated_at = NOW()
 		WHERE id = $1
-	`, task.ID, completedAt, skipReason, nullableString(req.ErrorMessage)); err != nil {
+	`, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage)); err != nil {
 		return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status")
 	}
 
@@ -978,12 +979,34 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 	completedAtText := completedAt.Format(time.RFC3339)
 	return &MonitoringSkipTaskResponse{
 		TaskID:      task.ID,
-		TaskStatus:  "skipped",
+		TaskStatus:  skipOutcome.TaskStatus,
 		SkipReason:  skipReason,
 		CompletedAt: &completedAtText,
 	}, nil
 }
 
+type monitoringSkipOutcome struct {
+	TaskStatus       string
+	StoredSkipReason string
+}
+
+func monitoringSkipOutcomeForReason(skipReason string) monitoringSkipOutcome {
+	switch strings.TrimSpace(skipReason) {
+	case "runtime_unknown":
+		return monitoringSkipOutcome{TaskStatus: "failed"}
+	default:
+		return monitoringSkipOutcome{TaskStatus: "skipped", StoredSkipReason: strings.TrimSpace(skipReason)}
+	}
+}
+
+func nullableStringPtr(value string) *string {
+	trimmed := strings.TrimSpace(value)
+	if trimmed == "" {
+		return nil
+	}
+	return &trimmed
+}
+
 func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
 	ctx context.Context,
 	client *repository.DesktopClient,
@@ -1504,7 +1527,7 @@ func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease(
 		FROM desktop_tasks
 		WHERE kind = 'monitor'
 		  AND monitor_task_id = $1
-		  AND status IN ('queued', 'in_progress', 'unknown')
+		  AND status IN ('queued', 'in_progress')
 		ORDER BY created_at DESC, id DESC
 		LIMIT 1
 	`, monitorTaskID).Scan(
@@ -2349,16 +2372,7 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
 		workspaceID     int64
 		activeAttemptID pgtype.UUID
 	)
-	if err := tx.QueryRow(ctx, `
-		SELECT desktop_id, workspace_id, active_attempt_id
-		FROM desktop_tasks
-		WHERE kind = 'monitor'
-		  AND monitor_task_id = $1
-		  AND status IN ('in_progress', 'unknown')
-		ORDER BY created_at DESC, id DESC
-		LIMIT 1
-		FOR UPDATE
-	`, task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil {
+	if err := tx.QueryRow(ctx, completeLinkedDesktopMonitorTaskLookupSQL(), task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil {
 		if err == pgx.ErrNoRows {
 			return nil
 		}
@@ -2434,6 +2448,19 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
 	return nil
 }
 
+func completeLinkedDesktopMonitorTaskLookupSQL() string {
+	return `
+		SELECT desktop_id, workspace_id, active_attempt_id
+		FROM desktop_tasks
+		WHERE kind = 'monitor'
+		  AND monitor_task_id = $1
+		  AND status = 'in_progress'
+		ORDER BY created_at DESC, id DESC
+		LIMIT 1
+		FOR UPDATE
+	`
+}
+
 func (s *MonitoringCallbackService) replaceCitationFacts(ctx context.Context, tx pgx.Tx, tenantID, runID int64) error {
 	if _, err := tx.Exec(ctx, `
 		DELETE FROM monitoring_citation_facts

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(
@@ -612,16 +569,7 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 		existingLeaseExpiresAt pgtype.Timestamptz
 		existingActiveAttempt  pgtype.UUID
 	)
-	err := tx.QueryRow(ctx, `
-		SELECT desktop_id, status, lease_expires_at, active_attempt_id
-		FROM desktop_tasks
-		WHERE kind = 'monitor'
-		  AND monitor_task_id = $1
-		  AND status IN ('queued', 'in_progress', 'unknown')
-		ORDER BY created_at DESC, id DESC
-		LIMIT 1
-		FOR UPDATE
-	`, spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
+	err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt)
 	switch err {
 	case nil:
 		task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID)
@@ -771,6 +719,19 @@ func (s *MonitoringService) upsertMonitorDesktopTask(
 	return task, true, false, nil
 }
 
+func upsertMonitorDesktopTaskActiveLookupSQL() string {
+	return `
+		SELECT desktop_id, status, lease_expires_at, active_attempt_id
+		FROM desktop_tasks
+		WHERE kind = 'monitor'
+		  AND monitor_task_id = $1
+		  AND status IN ('queued', 'in_progress')
+		ORDER BY created_at DESC, id DESC
+		LIMIT 1
+		FOR UPDATE
+	`
+}
+
 func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Context, taskIDs []int64) error {
 	if s == nil || s.monitoringPool == nil || len(taskIDs) == 0 {
 		return nil

server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go

@@ -179,19 +179,32 @@ func TestMarshalMonitorDesktopTaskPayloadUsesQuestionTextNotFallbackTitle(t *tes
 	assert.Equal(t, "通义千问 · 幽灵门轨道长度定制", decoded["title"])
 }
 
-func TestMonitorDesktopTaskPlatformSlotBusySQLThrottlesPerClientPlatformInProgress(t *testing.T) {
-	query := strings.Join(strings.Fields(monitorDesktopTaskPlatformSlotBusySQL()), " ")
+func TestUpsertMonitorDesktopTaskIgnoresUnknownAsActive(t *testing.T) {
+	query := strings.Join(strings.Fields(upsertMonitorDesktopTaskActiveLookupSQL()), " ")
 
-	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)
-	}
+	assert.Contains(t, query, "status IN ('queued', 'in_progress')")
+	assert.NotContains(t, query, "'unknown'")
+}
+
+func TestCompleteLinkedDesktopMonitorTaskOnlyFinalizesInProgress(t *testing.T) {
+	query := strings.Join(strings.Fields(completeLinkedDesktopMonitorTaskLookupSQL()), " ")
+
+	assert.Contains(t, query, "status = 'in_progress'")
+	assert.NotContains(t, query, "'unknown'")
+}
+
+func TestMonitoringSkipOutcomeForRuntimeUnknownFailsTask(t *testing.T) {
+	outcome := monitoringSkipOutcomeForReason("runtime_unknown")
+
+	assert.Equal(t, "failed", outcome.TaskStatus)
+	assert.Empty(t, outcome.StoredSkipReason)
+}
+
+func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) {
+	outcome := monitoringSkipOutcomeForReason("stale_business_date")
+
+	assert.Equal(t, "skipped", outcome.TaskStatus)
+	assert.Equal(t, "stale_business_date", outcome.StoredSkipReason)
 }
 
 func TestRequestPhase2MonitorInterruptsAcceptsExcludedPlatforms(t *testing.T) {

server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql

@@ -54,4 +54,4 @@ CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active
     ON desktop_tasks (monitor_task_id)
     WHERE kind = 'monitor'
       AND monitor_task_id IS NOT NULL
-      AND status IN ('queued', 'in_progress', 'unknown');
+      AND status IN ('queued', 'in_progress');

server/migrations/20260622152000_fail_monitor_unknown_tasks.down.sql

@@ -0,0 +1,7 @@
+DROP INDEX IF EXISTS uq_desktop_tasks_monitor_active;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active
+    ON desktop_tasks (monitor_task_id)
+    WHERE kind = 'monitor'
+      AND monitor_task_id IS NOT NULL
+      AND status IN ('queued', 'in_progress', 'unknown');

server/migrations/20260622152000_fail_monitor_unknown_tasks.up.sql

@@ -0,0 +1,13 @@
+UPDATE desktop_tasks
+SET status = 'failed',
+    updated_at = NOW()
+WHERE kind = 'monitor'
+  AND status = 'unknown';
+
+DROP INDEX IF EXISTS uq_desktop_tasks_monitor_active;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active
+    ON desktop_tasks (monitor_task_id)
+    WHERE kind = 'monitor'
+      AND monitor_task_id IS NOT NULL
+      AND status IN ('queued', 'in_progress');