fix collect-now active request reconciliation

server/internal/tenant/app/monitoring_collect_outbox_worker.go

@@ -616,13 +616,7 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats(
 		return stats, nil
 	}
 
-	if err := s.businessPool.QueryRow(ctx, `
+	if err := s.businessPool.QueryRow(ctx, collectNowActiveDesktopTaskCountSQL(), monitorTaskIDs).Scan(
-		SELECT COUNT(*)::bigint
-		FROM desktop_tasks d
-		WHERE d.kind = 'monitor'
-		  AND d.status = 'in_progress'
-		  AND d.monitor_task_id = ANY($1::bigint[])
-	`, monitorTaskIDs).Scan(
 		&stats.ActiveDesktopTaskCount,
 	); err != nil {
 		return stats, response.ErrInternal(50041, "query_failed", "failed to inspect collect-now desktop task progress")
@@ -631,6 +625,16 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats(
 	return stats, nil
 }
 
+func collectNowActiveDesktopTaskCountSQL() string {
+	return `
+		SELECT COUNT(*)::bigint
+		FROM desktop_tasks d
+		WHERE d.kind = 'monitor'
+		  AND d.status IN ('queued', 'in_progress')
+		  AND d.monitor_task_id = ANY($1::bigint[])
+	`
+}
+
 func (s *MonitoringService) loadCollectNowMonitorTaskIDs(
 	ctx context.Context,
 	request monitoringCollectRequestReconcileRow,
@@ -806,6 +810,9 @@ func deriveCollectNowRequestStatus(
 	if now.Before(request.TTLExpiresAt) {
 		return "", "", false
 	}
+	if stats.Leased > 0 || stats.Received > 0 || stats.ActiveDesktopTaskCount > 0 {
+		return "", "", false
+	}
 	if !request.FirstDispatchedAt.Valid {
 		return "timed_out", "立即采集请求超时，5 分钟内未能派发首条任务。", request.Status != "timed_out"
 	}

server/internal/tenant/app/monitoring_collect_outbox_worker_test.go

@@ -0,0 +1,40 @@
+package app
+
+import (
+	"database/sql"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDeriveCollectNowRequestStatusKeepsRunningWhenDesktopTaskActiveAfterTTL(t *testing.T) {
+	now := time.Date(2026, 6, 27, 1, 37, 30, 0, time.UTC)
+	request := monitoringCollectRequestReconcileRow{
+		Status:       "running",
+		TTLExpiresAt: now.Add(-30 * time.Second),
+		FirstDispatchedAt: sql.NullTime{
+			Time:  now.Add(-5 * time.Minute),
+			Valid: true,
+		},
+	}
+	stats := monitoringCollectRequestTaskStats{
+		Total:                  30,
+		Terminal:               6,
+		Completed:              6,
+		ActiveDesktopTaskCount: 1,
+	}
+
+	nextStatus, nextMessage, shouldFinalize := deriveCollectNowRequestStatus(request, stats, now)
+
+	assert.Empty(t, nextStatus)
+	assert.Empty(t, nextMessage)
+	assert.False(t, shouldFinalize)
+}
+
+func TestLoadCollectNowRequestTaskStatsTreatsQueuedDesktopTasksAsActive(t *testing.T) {
+	query := strings.Join(strings.Fields(collectNowActiveDesktopTaskCountSQL()), " ")
+
+	assert.Contains(t, query, "d.status IN ('queued', 'in_progress')")
+}